430 lines
14 KiB
Python
430 lines
14 KiB
Python
"""
|
|
Commands for profile-related operations.
|
|
"""
|
|
from typing import Any, Dict, Optional
|
|
from django.contrib.auth.models import User
|
|
|
|
from backend.core.models.profiles.profiles import Profile
|
|
from backend.core.repositories.profiles.profiles import ProfileRepository
|
|
from backend.core.utils.validators import (
|
|
is_valid_uuid, is_valid_email, is_valid_phone,
|
|
validate_required_fields
|
|
)
|
|
from backend.core.utils.helpers import generate_uuid
|
|
from backend.core.commands.base import Command, CommandResult
|
|
|
|
|
|
class CreateProfileCommand(Command):
|
|
"""
|
|
Command to create a new profile.
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
profile_repo: ProfileRepository,
|
|
user: User,
|
|
first_name: str,
|
|
last_name: str,
|
|
primary_phone: str,
|
|
email: str,
|
|
role: str = 'team_member',
|
|
secondary_phone: Optional[str] = None
|
|
):
|
|
"""
|
|
Initialize the create profile command.
|
|
|
|
Args:
|
|
profile_repo: Repository for profile operations.
|
|
user: The Django User object to associate with the profile.
|
|
first_name: First name of the profile.
|
|
last_name: Last name of the profile.
|
|
primary_phone: Primary phone number.
|
|
email: Email address.
|
|
role: Role of the profile ('admin', 'team_leader', 'team_member').
|
|
secondary_phone: Optional secondary phone number.
|
|
"""
|
|
self.profile_repo = profile_repo
|
|
self.user = user
|
|
self.first_name = first_name
|
|
self.last_name = last_name
|
|
self.primary_phone = primary_phone
|
|
self.email = email
|
|
self.role = role
|
|
self.secondary_phone = secondary_phone
|
|
|
|
def validate(self) -> Dict[str, Any]:
|
|
"""
|
|
Validate the profile creation data.
|
|
|
|
Returns:
|
|
Dict[str, Any]: Validation result with 'is_valid' and optional 'errors'.
|
|
"""
|
|
errors = []
|
|
|
|
# Check required fields
|
|
missing_fields = validate_required_fields(
|
|
{
|
|
'user': self.user,
|
|
'first_name': self.first_name,
|
|
'last_name': self.last_name,
|
|
'primary_phone': self.primary_phone,
|
|
'email': self.email
|
|
},
|
|
['user', 'first_name', 'last_name', 'primary_phone', 'email']
|
|
)
|
|
|
|
if missing_fields:
|
|
errors.append(f"Required fields missing: {', '.join(missing_fields)}")
|
|
|
|
# Validate email format
|
|
if not errors and self.email and not is_valid_email(self.email):
|
|
errors.append("Invalid email format.")
|
|
|
|
# Validate phone formats
|
|
if not errors and self.primary_phone and not is_valid_phone(self.primary_phone):
|
|
errors.append("Invalid primary phone format.")
|
|
|
|
if not errors and self.secondary_phone and not is_valid_phone(self.secondary_phone):
|
|
errors.append("Invalid secondary phone format.")
|
|
|
|
# Validate role
|
|
valid_roles = ['admin', 'team_leader', 'team_member']
|
|
if not errors and self.role not in valid_roles:
|
|
errors.append(f"Invalid role. Must be one of: {', '.join(valid_roles)}")
|
|
|
|
# Check if user already has a profile
|
|
if not errors and self.profile_repo.get_by_user(self.user):
|
|
errors.append(f"User already has a profile")
|
|
|
|
# Check if email is already in use
|
|
if not errors and self.profile_repo.get_by_email(self.email):
|
|
errors.append(f"Email is already in use")
|
|
|
|
return {
|
|
'is_valid': len(errors) == 0,
|
|
'errors': errors
|
|
}
|
|
|
|
def execute(self) -> CommandResult[Profile]:
|
|
"""
|
|
Execute the profile creation command.
|
|
|
|
Returns:
|
|
CommandResult[Profile]: Result of the command execution.
|
|
"""
|
|
# Validate command data
|
|
validation = self.validate()
|
|
if not validation['is_valid']:
|
|
return CommandResult.failure_result(validation['errors'])
|
|
|
|
try:
|
|
# Create profile data
|
|
profile_id = generate_uuid()
|
|
|
|
# Create profile data dictionary
|
|
profile_data = {
|
|
'id': profile_id,
|
|
'user': self.user,
|
|
'first_name': self.first_name,
|
|
'last_name': self.last_name,
|
|
'primary_phone': self.primary_phone,
|
|
'secondary_phone': self.secondary_phone,
|
|
'email': self.email,
|
|
'role': self.role
|
|
}
|
|
|
|
# Save to repository
|
|
created_profile = self.profile_repo.create(profile_data)
|
|
|
|
return CommandResult.success_result(
|
|
created_profile,
|
|
f"Profile for {self.first_name} {self.last_name} created successfully"
|
|
)
|
|
|
|
except Exception as e:
|
|
return CommandResult.failure_result(
|
|
str(e),
|
|
"Failed to create profile"
|
|
)
|
|
|
|
|
|
class UpdateProfileCommand(Command):
|
|
"""
|
|
Command to update an existing profile.
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
profile_repo: ProfileRepository,
|
|
profile_id: str,
|
|
first_name: Optional[str] = None,
|
|
last_name: Optional[str] = None,
|
|
primary_phone: Optional[str] = None,
|
|
email: Optional[str] = None,
|
|
role: Optional[str] = None,
|
|
secondary_phone: Optional[str] = None
|
|
):
|
|
"""
|
|
Initialize the update profile command.
|
|
|
|
Args:
|
|
profile_repo: Repository for profile operations.
|
|
profile_id: ID of the profile to update.
|
|
first_name: New first name.
|
|
last_name: New last name.
|
|
primary_phone: New primary phone number.
|
|
email: New email address.
|
|
role: New role.
|
|
secondary_phone: New secondary phone number.
|
|
"""
|
|
self.profile_repo = profile_repo
|
|
self.profile_id = profile_id
|
|
self.first_name = first_name
|
|
self.last_name = last_name
|
|
self.primary_phone = primary_phone
|
|
self.email = email
|
|
self.role = role
|
|
self.secondary_phone = secondary_phone
|
|
|
|
def validate(self) -> Dict[str, Any]:
|
|
"""
|
|
Validate the profile update data.
|
|
|
|
Returns:
|
|
Dict[str, Any]: Validation result with 'is_valid' and optional 'errors'.
|
|
"""
|
|
errors = []
|
|
|
|
# Validate profile exists
|
|
if not is_valid_uuid(self.profile_id):
|
|
errors.append("Invalid profile ID format")
|
|
else:
|
|
profile = self.profile_repo.get_by_id(self.profile_id)
|
|
if not profile:
|
|
errors.append(f"Profile with ID {self.profile_id} not found")
|
|
|
|
# Validate email format if provided
|
|
if not errors and self.email is not None and not is_valid_email(self.email):
|
|
errors.append("Invalid email format.")
|
|
|
|
# Validate phone formats if provided
|
|
if not errors and self.primary_phone is not None and not is_valid_phone(self.primary_phone):
|
|
errors.append("Invalid primary phone format.")
|
|
|
|
if not errors and self.secondary_phone is not None and not is_valid_phone(self.secondary_phone):
|
|
errors.append("Invalid secondary phone format.")
|
|
|
|
# Validate role if provided
|
|
valid_roles = ['admin', 'team_leader', 'team_member']
|
|
if not errors and self.role is not None and self.role not in valid_roles:
|
|
errors.append(f"Invalid role. Must be one of: {', '.join(valid_roles)}")
|
|
|
|
# Check if email is already in use by another profile
|
|
if not errors and self.email is not None:
|
|
existing_profile = self.profile_repo.get_by_email(self.email)
|
|
if existing_profile and str(existing_profile.id) != self.profile_id:
|
|
errors.append(f"Email is already in use by another profile")
|
|
|
|
return {
|
|
'is_valid': len(errors) == 0,
|
|
'errors': errors
|
|
}
|
|
|
|
def execute(self) -> CommandResult[Profile]:
|
|
"""
|
|
Execute the profile update command.
|
|
|
|
Returns:
|
|
CommandResult[Profile]: Result of the command execution.
|
|
"""
|
|
# Validate command data
|
|
validation = self.validate()
|
|
if not validation['is_valid']:
|
|
return CommandResult.failure_result(validation['errors'])
|
|
|
|
try:
|
|
# Create a dictionary of fields to update
|
|
update_data = {}
|
|
|
|
# Add fields to update_data if they were provided
|
|
if self.first_name is not None:
|
|
update_data['first_name'] = self.first_name
|
|
|
|
if self.last_name is not None:
|
|
update_data['last_name'] = self.last_name
|
|
|
|
if self.primary_phone is not None:
|
|
update_data['primary_phone'] = self.primary_phone
|
|
|
|
if self.secondary_phone is not None:
|
|
update_data['secondary_phone'] = self.secondary_phone
|
|
|
|
if self.email is not None:
|
|
update_data['email'] = self.email
|
|
|
|
if self.role is not None:
|
|
update_data['role'] = self.role
|
|
|
|
# Update the profile with the data dictionary
|
|
updated_profile = self.profile_repo.update(self.profile_id, update_data)
|
|
|
|
return CommandResult.success_result(
|
|
updated_profile,
|
|
f"Profile {self.profile_id} updated successfully"
|
|
)
|
|
|
|
except Exception as e:
|
|
return CommandResult.failure_result(
|
|
str(e),
|
|
"Failed to update profile"
|
|
)
|
|
|
|
|
|
class DeleteProfileCommand(Command):
|
|
"""
|
|
Command to delete a profile.
|
|
"""
|
|
|
|
def __init__(self, profile_repo: ProfileRepository, profile_id: str):
|
|
"""
|
|
Initialize the delete profile command.
|
|
|
|
Args:
|
|
profile_repo: Repository for profile operations.
|
|
profile_id: ID of the profile to delete.
|
|
"""
|
|
self.profile_repo = profile_repo
|
|
self.profile_id = profile_id
|
|
|
|
def validate(self) -> Dict[str, Any]:
|
|
"""
|
|
Validate the profile deletion request.
|
|
|
|
Returns:
|
|
Dict[str, Any]: Validation result with 'is_valid' and optional 'errors'.
|
|
"""
|
|
errors = []
|
|
|
|
# Validate profile exists
|
|
if not is_valid_uuid(self.profile_id):
|
|
errors.append("Invalid profile ID format")
|
|
else:
|
|
profile = self.profile_repo.get_by_id(self.profile_id)
|
|
if not profile:
|
|
errors.append(f"Profile with ID {self.profile_id} not found")
|
|
|
|
return {
|
|
'is_valid': len(errors) == 0,
|
|
'errors': errors
|
|
}
|
|
|
|
def execute(self) -> CommandResult[bool]:
|
|
"""
|
|
Execute the profile deletion command.
|
|
|
|
Returns:
|
|
CommandResult[bool]: Result of the command execution.
|
|
"""
|
|
# Validate command data
|
|
validation = self.validate()
|
|
if not validation['is_valid']:
|
|
return CommandResult.failure_result(validation['errors'])
|
|
|
|
try:
|
|
# Delete the profile
|
|
success = self.profile_repo.delete(self.profile_id)
|
|
|
|
if success:
|
|
return CommandResult.success_result(
|
|
True,
|
|
f"Profile {self.profile_id} deleted successfully"
|
|
)
|
|
else:
|
|
return CommandResult.failure_result(
|
|
"Failed to delete profile",
|
|
f"Profile {self.profile_id} could not be deleted"
|
|
)
|
|
|
|
except Exception as e:
|
|
return CommandResult.failure_result(
|
|
str(e),
|
|
"Failed to delete profile"
|
|
)
|
|
|
|
|
|
class SearchProfilesCommand(Command):
|
|
"""
|
|
Command to search for profiles.py.
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
profile_repo: ProfileRepository,
|
|
search_term: str,
|
|
role: Optional[str] = None
|
|
):
|
|
"""
|
|
Initialize the search profiles.py command.
|
|
|
|
Args:
|
|
profile_repo: Repository for profile operations.
|
|
search_term: The search term to look for.
|
|
role: Optional role to filter by.
|
|
"""
|
|
self.profile_repo = profile_repo
|
|
self.search_term = search_term
|
|
self.role = role
|
|
|
|
def validate(self) -> Dict[str, Any]:
|
|
"""
|
|
Validate the profile search request.
|
|
|
|
Returns:
|
|
Dict[str, Any]: Validation result with 'is_valid' and optional 'errors'.
|
|
"""
|
|
errors = []
|
|
|
|
# Validate role if provided
|
|
valid_roles = ['admin', 'team_leader', 'team_member']
|
|
if self.role is not None and self.role not in valid_roles:
|
|
errors.append(f"Invalid role. Must be one of: {', '.join(valid_roles)}")
|
|
|
|
return {
|
|
'is_valid': len(errors) == 0,
|
|
'errors': errors
|
|
}
|
|
|
|
def execute(self) -> CommandResult[list]:
|
|
"""
|
|
Execute the profile search command.
|
|
|
|
Returns:
|
|
CommandResult[list]: Result of the command execution with list of matching profiles.py.
|
|
"""
|
|
# Validate command data
|
|
validation = self.validate()
|
|
if not validation['is_valid']:
|
|
return CommandResult.failure_result(validation['errors'])
|
|
|
|
try:
|
|
# Search for profiles.py
|
|
results = self.profile_repo.search(self.search_term)
|
|
|
|
# Filter by role if provided
|
|
if self.role:
|
|
results = results.filter(role=self.role)
|
|
|
|
# Convert QuerySet to list
|
|
profiles = list(results)
|
|
|
|
return CommandResult.success_result(
|
|
profiles,
|
|
f"Found {len(profiles)} matching profiles.py"
|
|
)
|
|
|
|
except Exception as e:
|
|
return CommandResult.failure_result(
|
|
str(e),
|
|
"Failed to search profiles.py"
|
|
)
|