A R R R Associates
124 lines
6.6 KiB
Python
124 lines
6.6 KiB
Python
from __future__ import annotations
|
|
|
|
from sqlalchemy import exists, func, or_, select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.modules.client_groups.models import ClientGroup
|
|
from app.modules.clients.models import Client
|
|
from app.modules.services.models import ClientServiceSubscription
|
|
|
|
GROUP_TYPES = ("Family", "Business Group", "Promoter Group", "Trust Group", "Common Management", "Other")
|
|
|
|
|
|
def normalise_group_code(value: str | None) -> str:
|
|
return (value or "").strip().upper()
|
|
|
|
|
|
def list_groups(db: Session, *, tenant_id: int, include_inactive: bool = False, viewer_partner_id: int | None = None):
|
|
stmt = (
|
|
select(ClientGroup, func.count(Client.id).label("client_count"))
|
|
.outerjoin(Client, Client.client_group_id == ClientGroup.id)
|
|
.where(ClientGroup.tenant_id == tenant_id)
|
|
.group_by(ClientGroup.id)
|
|
.order_by(ClientGroup.group_name.asc())
|
|
)
|
|
if viewer_partner_id is not None:
|
|
review_access = exists(select(ClientServiceSubscription.id).where(
|
|
ClientServiceSubscription.tenant_id == tenant_id,
|
|
ClientServiceSubscription.client_id == Client.id,
|
|
ClientServiceSubscription.review_partner_user_id == viewer_partner_id,
|
|
ClientServiceSubscription.is_active.is_(True),
|
|
))
|
|
stmt = stmt.where(or_(Client.partner_id == viewer_partner_id, review_access))
|
|
if not include_inactive:
|
|
stmt = stmt.where(ClientGroup.is_active.is_(True))
|
|
return [{"group": group, "client_count": int(count or 0)} for group, count in db.execute(stmt).all()]
|
|
|
|
|
|
def get_group(db: Session, *, tenant_id: int, group_id: int):
|
|
return db.execute(select(ClientGroup).where(ClientGroup.id == group_id, ClientGroup.tenant_id == tenant_id)).scalar_one_or_none()
|
|
|
|
|
|
def get_group_by_code(db: Session, *, tenant_id: int, group_code: str):
|
|
code = normalise_group_code(group_code)
|
|
if not code:
|
|
return None
|
|
return db.execute(select(ClientGroup).where(ClientGroup.tenant_id == tenant_id, ClientGroup.group_code == code)).scalar_one_or_none()
|
|
|
|
|
|
def list_group_clients(db: Session, *, tenant_id: int, group_id: int, viewer_partner_id: int | None = None):
|
|
stmt = select(Client).where(Client.tenant_id == tenant_id, Client.client_group_id == group_id, Client.is_archived.is_(False))
|
|
if viewer_partner_id is not None:
|
|
review_access = exists(select(ClientServiceSubscription.id).where(
|
|
ClientServiceSubscription.tenant_id == tenant_id,
|
|
ClientServiceSubscription.client_id == Client.id,
|
|
ClientServiceSubscription.review_partner_user_id == viewer_partner_id,
|
|
ClientServiceSubscription.is_active.is_(True),
|
|
))
|
|
stmt = stmt.where(or_(Client.partner_id == viewer_partner_id, review_access))
|
|
return db.execute(stmt.order_by(Client.is_group_head.desc(), Client.client_name.asc())).scalars().all()
|
|
|
|
|
|
def create_group(db: Session, *, tenant_id: int, actor_user_id: int, payload: dict):
|
|
code = normalise_group_code(payload.get("group_code"))
|
|
name = (payload.get("group_name") or "").strip()
|
|
if not code or not name:
|
|
raise ValueError("Group code and group name are required.")
|
|
if get_group_by_code(db, tenant_id=tenant_id, group_code=code):
|
|
raise ValueError("Client group code already exists in this firm.")
|
|
group_type = (payload.get("group_type") or "Family").strip()
|
|
if group_type not in GROUP_TYPES:
|
|
raise ValueError("Invalid group type.")
|
|
row = ClientGroup(
|
|
tenant_id=tenant_id, group_code=code, group_name=name, group_type=group_type,
|
|
primary_contact_name=(payload.get("primary_contact_name") or "").strip() or None,
|
|
primary_contact_mobile=(payload.get("primary_contact_mobile") or "").strip() or None,
|
|
primary_contact_email=(payload.get("primary_contact_email") or "").strip().lower() or None,
|
|
assigned_partner_user_id=payload.get("assigned_partner_user_id") or None,
|
|
primary_consultant_id=payload.get("primary_consultant_id") or None,
|
|
notes=(payload.get("notes") or "").strip() or None,
|
|
is_active=bool(payload.get("is_active", True)), created_by_user_id=actor_user_id, updated_by_user_id=actor_user_id,
|
|
)
|
|
db.add(row); db.commit(); db.refresh(row); return row
|
|
|
|
|
|
def update_group(db: Session, *, row: ClientGroup, actor_user_id: int, payload: dict):
|
|
code = normalise_group_code(payload.get("group_code"))
|
|
name = (payload.get("group_name") or "").strip()
|
|
if not code or not name:
|
|
raise ValueError("Group code and group name are required.")
|
|
duplicate = get_group_by_code(db, tenant_id=row.tenant_id, group_code=code)
|
|
if duplicate and duplicate.id != row.id:
|
|
raise ValueError("Client group code already exists in this firm.")
|
|
group_type = (payload.get("group_type") or "Family").strip()
|
|
if group_type not in GROUP_TYPES:
|
|
raise ValueError("Invalid group type.")
|
|
for key, value in {
|
|
"group_code": code, "group_name": name, "group_type": group_type,
|
|
"primary_contact_name": (payload.get("primary_contact_name") or "").strip() or None,
|
|
"primary_contact_mobile": (payload.get("primary_contact_mobile") or "").strip() or None,
|
|
"primary_contact_email": (payload.get("primary_contact_email") or "").strip().lower() or None,
|
|
"assigned_partner_user_id": payload.get("assigned_partner_user_id") or None,
|
|
"primary_consultant_id": payload.get("primary_consultant_id") or None,
|
|
"notes": (payload.get("notes") or "").strip() or None,
|
|
"is_active": bool(payload.get("is_active", False)), "updated_by_user_id": actor_user_id,
|
|
}.items(): setattr(row, key, value)
|
|
db.add(row); db.commit(); db.refresh(row); return row
|
|
|
|
|
|
def resolve_or_create_group(db: Session, *, tenant_id: int, actor_user_id: int, group_code: str | None, group_name: str | None, group_type: str | None = None):
|
|
code = normalise_group_code(group_code)
|
|
name = (group_name or "").strip()
|
|
if not code and not name:
|
|
return None
|
|
if not code:
|
|
raise ValueError("client_group_code is required when client_group_name is supplied.")
|
|
existing = get_group_by_code(db, tenant_id=tenant_id, group_code=code)
|
|
if existing:
|
|
if name and existing.group_name.strip().lower() != name.lower():
|
|
raise ValueError(f"Client group code {code} already exists with name {existing.group_name}.")
|
|
return existing
|
|
if not name:
|
|
raise ValueError(f"Client group {code} does not exist; provide client_group_name to create it.")
|
|
return create_group(db, tenant_id=tenant_id, actor_user_id=actor_user_id, payload={"group_code": code, "group_name": name, "group_type": group_type or "Family", "is_active": True})
|