""" User repository for database operations on User models. """ import asyncio from datetime import UTC, datetime from typing import cast from uuid import UUID from sqlalchemy import ColumnElement, func, select from sqlalchemy.ext.asyncio import AsyncSession from app.core.security import hash_password from app.models.user import RoleStr, User, UserRole from app.schemas.user import UserCreate, UserUpdate class UserRepository: """Repository handling all database queries for User entities.""" async def get_by_id(self, db: AsyncSession, user_id: UUID) -> User | None: """Fetch an active, non-deleted user by UUID.""" stmt = select(User).where(User.id == user_id, User.deleted_at.is_(None)) result = await db.execute(stmt) return result.scalars().first() async def get_by_phone( self, db: AsyncSession, phone: str, lab_id: UUID | None = None, ) -> User | None: """ Fetch a non-deleted user by phone number. If lab_id is provided, scopes the search to that specific lab branch. """ stmt = select(User).where(User.phone == phone, User.deleted_at.is_(None)) if lab_id is not None: stmt = stmt.where(User.lab_id == lab_id) result = await db.execute(stmt) return result.scalars().first() async def get_by_phone_including_deleted( self, db: AsyncSession, phone: str, lab_id: UUID | None = None, ) -> User | None: """Fetch a user by phone including soft-deleted rows (for safe re-hire).""" stmt = select(User).where(User.phone == phone) if lab_id is not None: stmt = stmt.where(User.lab_id == lab_id) result = await db.execute(stmt) return result.scalars().first() async def restore_from_create(self, db: AsyncSession, user: User, data: UserCreate) -> User: """Re-activate a soft-deleted user and apply the new-hire payload.""" user.deleted_at = None user.name = data.name user.phone = data.phone user.email = data.email user.role = cast(RoleStr, data.role) user.lab_id = data.lab_id user.hashed_password = await asyncio.to_thread(hash_password, data.password) user.license_no = data.license_no user.commission_rate = data.commission_rate user.is_active = True await db.flush() await db.refresh(user) return user async def get_by_identifier( self, db: AsyncSession, identifier: str, lab_id: UUID | None = None, ) -> User | None: """ Fetch an active, non-deleted user by phone number or email address. """ stmt = select(User).where( (User.phone == identifier) | (User.email == identifier), User.deleted_at.is_(None), ) if lab_id is not None: stmt = stmt.where(User.lab_id == lab_id) result = await db.execute(stmt) return result.scalars().first() async def get_by_email( self, db: AsyncSession, email: str, lab_id: UUID, ) -> User | None: """Fetch a non-deleted user by email within a lab branch.""" stmt = select(User).where( User.email == email, User.lab_id == lab_id, User.deleted_at.is_(None), ) result = await db.execute(stmt) return result.scalars().first() async def get_by_email_ci( self, db: AsyncSession, email: str, ) -> User | None: """Fetch a non-deleted user by case-insensitive email address across all branches.""" stmt = select(User).where( func.lower(User.email) == email.strip().lower(), User.deleted_at.is_(None), ) result = await db.execute(stmt) return result.scalars().first() async def list_users( self, db: AsyncSession, limit: int = 50, offset: int = 0, lab_id: UUID | None = None, role: UserRole | None = None, is_active: bool | None = None, ) -> tuple[list[User], int]: """Fetch paginated staff users with optional filters.""" base_filters: list[ColumnElement[bool]] = [User.deleted_at.is_(None)] if lab_id is not None: base_filters.append(User.lab_id == lab_id) if role is not None: base_filters.append(User.role == role) if is_active is not None: base_filters.append(User.is_active == is_active) count_stmt = select(func.count()).select_from(User).where(*base_filters) total: int = (await db.execute(count_stmt)).scalar_one() data_stmt = ( select(User).where(*base_filters).order_by(User.created_at.desc()).offset(offset).limit(limit) ) res = await db.execute(data_stmt) return list(res.scalars().all()), total async def create(self, db: AsyncSession, data: UserCreate, hashed_password: str) -> User: """Create and persist a new staff user.""" user = User( lab_id=data.lab_id, name=data.name, phone=data.phone, email=data.email, role=data.role, hashed_password=hashed_password, is_active=True, license_no=data.license_no, commission_rate=data.commission_rate, ) db.add(user) await db.flush() await db.refresh(user) return user async def update(self, db: AsyncSession, user: User, data: UserUpdate) -> User: """Update fields of an existing user.""" if data.name is not None: user.name = data.name if data.phone is not None: user.phone = data.phone if data.email is not None: user.email = data.email if data.role is not None: user.role = cast(RoleStr, data.role) if data.commission_rate is not None: user.commission_rate = data.commission_rate if data.license_no is not None: user.license_no = data.license_no or None if data.is_active is not None: user.is_active = data.is_active await db.flush() await db.refresh(user) return user async def update_password(self, db: AsyncSession, user: User, new_hashed_password: str) -> None: """Update password hash for a user.""" user.hashed_password = new_hashed_password await db.flush() async def set_active_status(self, db: AsyncSession, user: User, is_active: bool) -> User: """Activate or deactivate user account.""" user.is_active = is_active await db.flush() await db.refresh(user) return user async def delete(self, db: AsyncSession, user: User) -> None: """Soft-delete a user record.""" user.deleted_at = datetime.now(UTC) await db.flush() async def count_all(self, db: AsyncSession) -> int: """Return the total number of non-deleted users in the system.""" stmt = select(func.count()).select_from(User).where(User.deleted_at.is_(None)) result = await db.execute(stmt) return int(result.scalar_one() or 0)