175 lines
5.0 KiB
Python
175 lines
5.0 KiB
Python
"""User management service."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import uuid
|
|
from typing import Any
|
|
|
|
from sqlalchemy import func, or_, select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.core.auth import hash_password
|
|
from app.models.user import User, UserTenant
|
|
|
|
# Sentinel used to distinguish "not provided" from "explicitly set to None".
|
|
# When ``role_id`` is _UNSET the service leaves the existing value untouched.
|
|
# When ``role_id`` is None the service clears the FK (falls back to role string).
|
|
_UNSET: Any = object()
|
|
|
|
|
|
class UserService:
|
|
"""Handles user CRUD operations."""
|
|
|
|
async def list_users(
|
|
self,
|
|
db: AsyncSession,
|
|
tenant_id: uuid.UUID,
|
|
page: int = 1,
|
|
page_size: int = 25,
|
|
search: str | None = None,
|
|
) -> dict[str, Any]:
|
|
"""List users in a tenant with pagination and search."""
|
|
offset = (page - 1) * page_size
|
|
|
|
q = select(User).where(User.tenant_id == tenant_id)
|
|
count_q = select(func.count()).select_from(User).where(User.tenant_id == tenant_id)
|
|
|
|
if search:
|
|
search_filter = or_(
|
|
User.name.ilike(f"%{search}%"),
|
|
User.email.ilike(f"%{search}%"),
|
|
)
|
|
q = q.where(search_filter)
|
|
count_q = count_q.where(search_filter)
|
|
|
|
total = (await db.execute(count_q)).scalar() or 0
|
|
|
|
q = q.offset(offset).limit(page_size).order_by(User.created_at.desc())
|
|
result = await db.execute(q)
|
|
users = result.scalars().all()
|
|
|
|
return {
|
|
"items": [self._user_to_dict(u) for u in users],
|
|
"total": total,
|
|
"page": page,
|
|
"page_size": page_size,
|
|
}
|
|
|
|
async def get_user(
|
|
self,
|
|
db: AsyncSession,
|
|
tenant_id: uuid.UUID,
|
|
user_id: uuid.UUID,
|
|
) -> User | None:
|
|
"""Get a single user by ID within tenant scope."""
|
|
q = select(User).where(User.id == user_id, User.tenant_id == tenant_id)
|
|
result = await db.execute(q)
|
|
return result.scalar_one_or_none()
|
|
|
|
async def create_user(
|
|
self,
|
|
db: AsyncSession,
|
|
tenant_id: uuid.UUID,
|
|
email: str,
|
|
name: str,
|
|
password: str,
|
|
role: str = "viewer",
|
|
role_id: uuid.UUID | None = None,
|
|
is_active: bool = True,
|
|
) -> User:
|
|
"""Create a new user in a tenant.
|
|
|
|
If role_id is provided it links the user to a custom Role record.
|
|
The legacy ``role`` string is kept for backward compatibility.
|
|
"""
|
|
user = User(
|
|
tenant_id=tenant_id,
|
|
email=email,
|
|
name=name,
|
|
password_hash=hash_password(password),
|
|
role=role,
|
|
role_id=role_id,
|
|
is_active=is_active,
|
|
preferences={},
|
|
)
|
|
db.add(user)
|
|
await db.flush()
|
|
|
|
# Add user-tenant membership
|
|
ut = UserTenant(
|
|
user_id=user.id,
|
|
tenant_id=tenant_id,
|
|
is_default=True,
|
|
)
|
|
db.add(ut)
|
|
await db.flush()
|
|
|
|
return user
|
|
|
|
async def update_user(
|
|
self,
|
|
db: AsyncSession,
|
|
tenant_id: uuid.UUID,
|
|
user_id: uuid.UUID,
|
|
name: str | None = None,
|
|
role: str | None = None,
|
|
role_id: uuid.UUID | None | Any = _UNSET,
|
|
is_active: bool | None = None,
|
|
) -> User | None:
|
|
"""Update a user.
|
|
|
|
``role_id`` uses a sentinel to distinguish three states:
|
|
- ``_UNSET`` (default): leave the existing role_id unchanged
|
|
- ``None``: clear the FK (fall back to the legacy ``role`` string)
|
|
- ``uuid.UUID``: link to a custom Role record
|
|
"""
|
|
q = select(User).where(User.id == user_id, User.tenant_id == tenant_id)
|
|
result = await db.execute(q)
|
|
user = result.scalar_one_or_none()
|
|
if user is None:
|
|
return None
|
|
|
|
if name is not None:
|
|
user.name = name
|
|
if role is not None:
|
|
user.role = role
|
|
if role_id is not _UNSET:
|
|
user.role_id = role_id
|
|
if is_active is not None:
|
|
user.is_active = is_active
|
|
|
|
await db.flush()
|
|
return user
|
|
|
|
async def delete_user(
|
|
self,
|
|
db: AsyncSession,
|
|
tenant_id: uuid.UUID,
|
|
user_id: uuid.UUID,
|
|
) -> bool:
|
|
"""Delete a user from a tenant."""
|
|
q = select(User).where(User.id == user_id, User.tenant_id == tenant_id)
|
|
result = await db.execute(q)
|
|
user = result.scalar_one_or_none()
|
|
if user is None:
|
|
return False
|
|
|
|
await db.delete(user)
|
|
await db.flush()
|
|
return True
|
|
|
|
def _user_to_dict(self, user: User) -> dict[str, Any]:
|
|
"""Convert user to response dict."""
|
|
return {
|
|
"id": str(user.id),
|
|
"email": user.email,
|
|
"name": user.name,
|
|
"role": user.role,
|
|
"role_id": str(user.role_id) if user.role_id else None,
|
|
"is_active": user.is_active,
|
|
"tenant_id": str(user.tenant_id),
|
|
}
|
|
|
|
|
|
user_service = UserService()
|