A R R R Associates
133 lines
4.0 KiB
Python
133 lines
4.0 KiB
Python
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,
|
|
} |