from __future__ import annotations from dataclasses import dataclass from math import ceil from sqlalchemy import or_, select from sqlalchemy.orm import Session from app.modules.core.iam.models import User from app.modules.core.iam.scope import ( UserScope, get_manageable_roles, get_user_role_names, list_visible_branches, list_visible_tenants, ) from app.modules.core.tenancy.models import Branch, Tenant @dataclass class PageResult: items: list page: int per_page: int total: int pages: int @property def has_prev(self) -> bool: return self.page > 1 @property def has_next(self) -> bool: return self.page < self.pages def paginate_list(items: list, page: int = 1, per_page: int = 10) -> PageResult: total = len(items) per_page = max(1, min(per_page, 100)) pages = max(1, ceil(total / per_page)) if total else 1 page = max(1, min(page, pages)) start = (page - 1) * per_page end = start + per_page return PageResult(items=items[start:end], page=page, per_page=per_page, total=total, pages=pages) def search_scoped_users(db: Session, scope: UserScope, q: str | None = None) -> list[User]: stmt = select(User) if not scope.is_system_admin: stmt = stmt.where(User.tenant_id == scope.actor.tenant_id) if scope.branch_scoped: stmt = stmt.where(User.branch_id == scope.actor.branch_id) query = (q or "").strip() if query: like = f"%{query}%" stmt = stmt.where(or_(User.email.ilike(like), User.full_name.ilike(like))) return db.execute(stmt.order_by(User.full_name, User.email, User.id)).scalars().all() def build_user_listing_payload( db: Session, scope: UserScope, q: str | None = None, page: int = 1, per_page: int = 10, ) -> dict: users = search_scoped_users(db, scope, q=q) paged = paginate_list(users, page=page, per_page=per_page) user_ids = [u.id for u in paged.items] roles = {user_id: get_user_role_names(db, user_id) for user_id in user_ids} tenants = {t.id: t for t in list_visible_tenants(db, scope)} if scope.is_system_admin: branch_rows = db.execute(select(Branch).order_by(Branch.name)).scalars().all() else: branch_rows = list_visible_branches(db, scope, scope.actor.tenant_id) branches = {b.id: b for b in branch_rows} return { "users_page": paged, "users": paged.items, "user_roles": roles, "tenants": tenants, "branches": branches, "filters": { "q": (q or "").strip(), "per_page": paged.per_page, }, } def build_user_form_payload( db: Session, scope: UserScope, actor: User, user_obj: User | None = None, assigned_role_ids: list[int] | None = None, ) -> dict: tenants = list_visible_tenants(db, scope) if user_obj: selected_tenant_id = user_obj.tenant_id selected_branch_id = user_obj.branch_id else: if scope.is_system_admin: selected_tenant_id = actor.tenant_id or (tenants[0].id if tenants else None) else: selected_tenant_id = actor.tenant_id selected_branch_id = actor.branch_id if scope.branch_scoped else None branches = list_visible_branches(db, scope, selected_tenant_id) selected_tenant = next((t for t in tenants if t.id == selected_tenant_id), None) selected_branch = next((b for b in branches if b.id == selected_branch_id), None) manageable_roles = get_manageable_roles(db, scope) return { "user_obj": user_obj, "assigned_role_ids": assigned_role_ids or [], "roles": manageable_roles, "tenants": tenants, "branches": branches, "scope": scope, "selected_tenant_id": selected_tenant_id, "selected_branch_id": selected_branch_id, "selected_tenant_name": selected_tenant.name if selected_tenant else "", "selected_branch_name": selected_branch.name if selected_branch else "", "can_change_tenant": scope.is_system_admin, }