410 lines
13 KiB
Python
410 lines
13 KiB
Python
"""
|
|
User Service - Business Logic for User Management
|
|
Handles user CRUD operations, profile management, and user administration
|
|
"""
|
|
from typing import Optional, Dict, Any, List
|
|
from app.core.database import db
|
|
from models import User, AuditLog
|
|
from app.repositories import UserRepository
|
|
import json
|
|
|
|
|
|
class UserService:
|
|
"""
|
|
User service - contains ALL business logic for user management.
|
|
|
|
Following Python Quick Start Guide:
|
|
- Service layer contains business rules
|
|
- Uses repository layer for database operations
|
|
- No HTTP/request handling (that stays in endpoints)
|
|
"""
|
|
|
|
def __init__(self, db_session=None):
|
|
"""Initialize user service with database session."""
|
|
self.db = db_session or db.session
|
|
self.user_repo = UserRepository(db_session)
|
|
|
|
def get_user_by_id(self, user_id: int) -> Optional[User]:
|
|
"""
|
|
Get user by ID.
|
|
|
|
Business Rules:
|
|
1. User must exist
|
|
2. Return None if not found (don't expose deleted users)
|
|
|
|
Args:
|
|
user_id: User's ID
|
|
|
|
Returns:
|
|
User object or None
|
|
"""
|
|
user = User.query.get(user_id)
|
|
return user if user else None
|
|
|
|
def get_all_users(self, page: int = 1, per_page: int = 50) -> Dict[str, Any]:
|
|
"""
|
|
Get all users with pagination.
|
|
|
|
Args:
|
|
page: Page number (1-indexed)
|
|
per_page: Items per page
|
|
|
|
Returns:
|
|
Dict with 'users' list and pagination info
|
|
"""
|
|
pagination = User.query.order_by(User.id.desc()).paginate(
|
|
page=page,
|
|
per_page=per_page,
|
|
error_out=False
|
|
)
|
|
|
|
return {
|
|
'users': pagination.items,
|
|
'total': pagination.total,
|
|
'page': pagination.page,
|
|
'per_page': pagination.per_page,
|
|
'pages': pagination.pages
|
|
}
|
|
|
|
def get_user_statistics(self) -> Dict[str, int]:
|
|
"""
|
|
Get user statistics.
|
|
|
|
Returns:
|
|
Dict with counts for total, active, inactive, and admin users
|
|
"""
|
|
total_users = User.query.count()
|
|
active_users = User.query.filter_by(is_active=True).count()
|
|
inactive_users = total_users - active_users
|
|
admin_users = User.query.filter_by(is_admin=True).count()
|
|
|
|
return {
|
|
'total_users': total_users,
|
|
'active_users': active_users,
|
|
'inactive_users': inactive_users,
|
|
'admin_users': admin_users
|
|
}
|
|
|
|
def create_user(
|
|
self,
|
|
username: str,
|
|
email: str,
|
|
name: str,
|
|
password: str,
|
|
role: str = 'user',
|
|
permissions_str: str = '',
|
|
is_admin: bool = False,
|
|
is_active: bool = True,
|
|
admin_id: Optional[int] = None,
|
|
ip_address: Optional[str] = None,
|
|
user_agent: Optional[str] = None
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Create a new user (admin operation).
|
|
|
|
Business Rules:
|
|
1. All required fields must be provided
|
|
2. Username must be unique
|
|
3. Email must be unique
|
|
4. Permissions must be valid JSON
|
|
5. Default to non-admin active user
|
|
6. Log creation event if admin_id provided
|
|
|
|
Args:
|
|
username: User's username
|
|
email: User's email
|
|
name: User's full name
|
|
password: User's password
|
|
role: User's role (default: 'user')
|
|
permissions_str: JSON string of permissions
|
|
is_admin: Whether user is an admin
|
|
is_active: Whether user is active
|
|
admin_id: ID of admin creating this user (for audit log)
|
|
ip_address: IP address for audit log
|
|
user_agent: User agent for audit log
|
|
|
|
Returns:
|
|
Dict with 'success' (bool), 'user_id' (if successful), or 'error'
|
|
"""
|
|
# Business Rule 1: Required fields
|
|
if not all([username, email, name, password]):
|
|
return {'success': False, 'error': 'All fields are required'}
|
|
|
|
# Business Rule 2: Username uniqueness
|
|
if User.query.filter_by(username=username).first():
|
|
return {'success': False, 'error': 'Username already exists'}
|
|
|
|
# Business Rule 3: Email uniqueness
|
|
if User.query.filter_by(email=email).first():
|
|
return {'success': False, 'error': 'Email already exists'}
|
|
|
|
# Business Rule 4: Parse permissions (comma-separated or JSON)
|
|
try:
|
|
if permissions_str:
|
|
# Try JSON first
|
|
try:
|
|
permissions = json.loads(permissions_str)
|
|
except json.JSONDecodeError:
|
|
# Fall back to comma-separated
|
|
permissions = [p.strip() for p in permissions_str.split(',') if p.strip()]
|
|
else:
|
|
permissions = []
|
|
except Exception:
|
|
return {'success': False, 'error': 'Invalid permissions format'}
|
|
|
|
# Create user
|
|
user = User(
|
|
username=username,
|
|
email=email,
|
|
name=name,
|
|
preferred_username=username,
|
|
role=role,
|
|
permissions=json.dumps(permissions) if permissions else None,
|
|
is_admin=is_admin,
|
|
is_active=is_active
|
|
)
|
|
user.set_password(password)
|
|
|
|
try:
|
|
self.db.add(user)
|
|
self.db.commit()
|
|
|
|
# Business Rule 6: Log creation if admin_id provided
|
|
if admin_id:
|
|
admin_user = User.query.get(admin_id)
|
|
if admin_user:
|
|
AuditLog.log(
|
|
action='user_created',
|
|
username=admin_user.username,
|
|
user_id=admin_user.id,
|
|
ip_address=ip_address,
|
|
user_agent=user_agent,
|
|
details={
|
|
'created_user': username,
|
|
'created_user_id': user.id,
|
|
'role': role,
|
|
'is_admin': is_admin
|
|
}
|
|
)
|
|
|
|
return {
|
|
'success': True,
|
|
'user_id': user.id,
|
|
'message': f'User "{username}" created successfully'
|
|
}
|
|
except Exception as e:
|
|
self.db.rollback()
|
|
return {'success': False, 'error': f'Failed to create user: {str(e)}'}
|
|
|
|
def update_user(
|
|
self,
|
|
user_id: int,
|
|
username: Optional[str] = None,
|
|
email: Optional[str] = None,
|
|
name: Optional[str] = None,
|
|
role: Optional[str] = None,
|
|
permissions_str: Optional[str] = None,
|
|
is_admin: Optional[bool] = None,
|
|
is_active: Optional[bool] = None,
|
|
new_password: Optional[str] = None
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Update an existing user.
|
|
|
|
Business Rules:
|
|
1. User must exist
|
|
2. If username changes, new username must be unique
|
|
3. If email changes, new email must be unique
|
|
4. Permissions must be valid (JSON or comma-separated) if provided
|
|
5. Update password if provided
|
|
|
|
Args:
|
|
user_id: ID of user to update
|
|
username: New username (optional)
|
|
email: New email (optional)
|
|
name: New name (optional)
|
|
role: New role (optional)
|
|
permissions_str: New permissions (JSON or comma-separated) (optional)
|
|
is_admin: New admin status (optional)
|
|
is_active: New active status (optional)
|
|
new_password: New password (optional)
|
|
|
|
Returns:
|
|
Dict with 'success' (bool) and 'message' or 'error'
|
|
"""
|
|
# Business Rule 1: User must exist
|
|
user = User.query.get(user_id)
|
|
if not user:
|
|
return {'success': False, 'error': 'User not found'}
|
|
|
|
# Business Rule 2: Username uniqueness (if changing)
|
|
if username and username != user.username:
|
|
if User.query.filter_by(username=username).first():
|
|
return {'success': False, 'error': 'Username already exists'}
|
|
user.username = username
|
|
|
|
# Business Rule 3: Email uniqueness (if changing)
|
|
if email and email != user.email:
|
|
if User.query.filter_by(email=email).first():
|
|
return {'success': False, 'error': 'Email already exists'}
|
|
user.email = email
|
|
|
|
# Update other fields if provided
|
|
if name:
|
|
user.name = name
|
|
if role:
|
|
user.role = role
|
|
|
|
# Business Rule 4: Parse permissions if provided (JSON or comma-separated)
|
|
if permissions_str is not None:
|
|
try:
|
|
if permissions_str:
|
|
# Try JSON first
|
|
try:
|
|
permissions = json.loads(permissions_str)
|
|
except json.JSONDecodeError:
|
|
# Fall back to comma-separated
|
|
permissions = [p.strip() for p in permissions_str.split(',') if p.strip()]
|
|
user.set_permissions(permissions)
|
|
else:
|
|
user.permissions = None
|
|
except Exception:
|
|
return {'success': False, 'error': 'Invalid permissions format'}
|
|
|
|
if is_admin is not None:
|
|
user.is_admin = is_admin
|
|
if is_active is not None:
|
|
user.is_active = is_active
|
|
|
|
# Business Rule 5: Update password if provided
|
|
if new_password:
|
|
user.set_password(new_password)
|
|
|
|
try:
|
|
self.db.commit()
|
|
return {
|
|
'success': True,
|
|
'message': f'User "{user.username}" updated successfully'
|
|
}
|
|
except Exception as e:
|
|
self.db.rollback()
|
|
return {'success': False, 'error': f'Failed to update user: {str(e)}'}
|
|
|
|
def deactivate_user(self, user_id: int) -> Dict[str, Any]:
|
|
"""
|
|
Deactivate a user.
|
|
|
|
Business Rules:
|
|
1. User must exist
|
|
2. Set is_active to False
|
|
|
|
Args:
|
|
user_id: ID of user to deactivate
|
|
|
|
Returns:
|
|
Dict with 'success' (bool) and 'message' or 'error'
|
|
"""
|
|
user = User.query.get(user_id)
|
|
if not user:
|
|
return {'success': False, 'error': 'User not found'}
|
|
|
|
user.is_active = False
|
|
|
|
try:
|
|
self.db.commit()
|
|
return {'success': True, 'message': f'User "{user.username}" deactivated'}
|
|
except Exception as e:
|
|
self.db.rollback()
|
|
return {'success': False, 'error': f'Failed to deactivate user: {str(e)}'}
|
|
|
|
def activate_user(self, user_id: int) -> Dict[str, Any]:
|
|
"""
|
|
Activate a user.
|
|
|
|
Business Rules:
|
|
1. User must exist
|
|
2. Set is_active to True
|
|
|
|
Args:
|
|
user_id: ID of user to activate
|
|
|
|
Returns:
|
|
Dict with 'success' (bool) and 'message' or 'error'
|
|
"""
|
|
user = User.query.get(user_id)
|
|
if not user:
|
|
return {'success': False, 'error': 'User not found'}
|
|
|
|
user.is_active = True
|
|
|
|
try:
|
|
self.db.commit()
|
|
return {'success': True, 'message': f'User "{user.username}" activated'}
|
|
except Exception as e:
|
|
self.db.rollback()
|
|
return {'success': False, 'error': f'Failed to activate user: {str(e)}'}
|
|
|
|
def delete_user(
|
|
self,
|
|
user_id: int,
|
|
admin_id: Optional[int] = None,
|
|
ip_address: Optional[str] = None,
|
|
user_agent: Optional[str] = None
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Delete a user.
|
|
|
|
Business Rules:
|
|
1. User must exist
|
|
2. Cannot delete last admin user
|
|
3. Permanently remove from database
|
|
4. Log deletion if admin_id provided
|
|
|
|
Args:
|
|
user_id: ID of user to delete
|
|
admin_id: ID of admin deleting the user (for audit log)
|
|
ip_address: IP address for audit log
|
|
user_agent: User agent for audit log
|
|
|
|
Returns:
|
|
Dict with 'success' (bool) and 'message' or 'error'
|
|
"""
|
|
user = User.query.get(user_id)
|
|
if not user:
|
|
return {'success': False, 'error': 'User not found'}
|
|
|
|
# Business Rule 2: Check if deleting last admin
|
|
if user.is_admin:
|
|
admin_count = User.query.filter_by(is_admin=True).count()
|
|
if admin_count <= 1:
|
|
return {'success': False, 'error': 'Cannot delete last admin user'}
|
|
|
|
# Save info for logging
|
|
username = user.username
|
|
deleted_user_id = user.id
|
|
|
|
try:
|
|
self.db.delete(user)
|
|
self.db.commit()
|
|
|
|
# Business Rule 4: Log deletion if admin_id provided
|
|
if admin_id:
|
|
admin_user = User.query.get(admin_id)
|
|
if admin_user:
|
|
AuditLog.log(
|
|
action='user_deleted',
|
|
username=admin_user.username,
|
|
user_id=admin_user.id,
|
|
ip_address=ip_address,
|
|
user_agent=user_agent,
|
|
details={
|
|
'deleted_user': username,
|
|
'deleted_user_id': deleted_user_id
|
|
}
|
|
)
|
|
|
|
return {'success': True, 'message': f'User "{username}" deleted permanently'}
|
|
except Exception as e:
|
|
self.db.rollback()
|
|
return {'success': False, 'error': f'Failed to delete user: {str(e)}'}
|