Rakseeibra/arrr-erp
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Enforce partner ownership on partner dashboard

Browse Source
This commit is contained in:
A R R R Associates
2026-07-29 15:52:25 +05:30
parent a58ec7b807
commit cc5a936d4e
2 changed files with 63 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/modules/partner_dashboard/service.py
+61 -9
View File
@@ -7,6 +7,7 @@ from typing import Any
from sqlalchemy import func, or_, select
from sqlalchemy.orm import Session, selectinload
from app.modules.clients.association_models import ClientAssociation
from app.modules.clients.models import Client
from app.modules.alerts.workflow_escalations import list_workflow_escalations
from app.modules.core.iam.models import User
@@ -89,6 +90,41 @@ def _financial_year(request) -> str | None:
return value or None
def _is_scoped_partner(roles: set[str]) -> bool:
"""Return True only for a normal Partner login.
System Admin and Firm Admin retain their existing administrative scope. A
normal Partner is always ownership-scoped, even when a branch is selected.
"""
return "Partner" in roles and "System Admin" not in roles and "Firm Admin" not in roles
def _partner_client_access_expression(*, partner_user_id: int, tenant_id: int | None):
association_access = exists(
select(ClientAssociation.id).where(
ClientAssociation.client_id == Client.id,
ClientAssociation.partner_user_id == int(partner_user_id),
or_(
ClientAssociation.firm_tenant_id == tenant_id,
ClientAssociation.firm_tenant_id.is_(None),
) if tenant_id is not None else ClientAssociation.id.is_not(None),
)
)
review_access = exists(
select(ClientServiceSubscription.id).where(
ClientServiceSubscription.client_id == Client.id,
ClientServiceSubscription.review_partner_user_id == int(partner_user_id),
ClientServiceSubscription.is_active.is_(True),
ClientServiceSubscription.tenant_id == tenant_id if tenant_id is not None else ClientServiceSubscription.id.is_not(None),
)
)
return or_(
Client.partner_id == int(partner_user_id),
association_access,
review_access,
)
def _task_scope(stmt, tenant_id: int | None, branch_id: int | None, current_user, roles: set[str], fy: str | None = None):
if tenant_id:
stmt = stmt.where(ClientServiceTaskInstance.tenant_id == tenant_id)
@@ -97,7 +133,7 @@ def _task_scope(stmt, tenant_id: int | None, branch_id: int | None, current_user
stmt = stmt.where(ClientServiceTaskInstance.branch_id == branch_id)
if fy:
stmt = stmt.where(ClientServiceTaskInstance.financial_year == fy)
if "Partner" in roles and "System Admin" not in roles and branch_id is None:
if _is_scoped_partner(roles):
stmt = stmt.where(
ClientServiceTaskInstance.subscription.has(
or_(
@@ -117,7 +153,7 @@ def _subscription_scope(stmt, tenant_id: int | None, branch_id: int | None, curr
stmt = stmt.where(ClientServiceSubscription.branch_id == branch_id)
if fy:
stmt = stmt.where(ClientServiceSubscription.financial_year == fy)
if "Partner" in roles and "System Admin" not in roles and branch_id is None:
if _is_scoped_partner(roles):
stmt = stmt.where(
or_(
ClientServiceSubscription.assigned_partner_user_id == current_user.id,
@@ -133,11 +169,11 @@ def _client_scope(stmt, tenant_id: int | None, branch_id: int | None, current_us
stmt = stmt.where(Client.is_active.is_(True))
if branch_id is not None:
stmt = stmt.where(Client.branch_id == branch_id)
elif "Partner" in roles and "System Admin" not in roles:
if _is_scoped_partner(roles):
stmt = stmt.where(
or_(
Client.partner_id == current_user.id,
Client.default_review_partner_user_id == current_user.id,
_partner_client_access_expression(
partner_user_id=int(current_user.id),
tenant_id=tenant_id,
)
)
return stmt
@@ -213,7 +249,7 @@ def _load_clients(db: Session, tenant_id: int | None, branch_id: int | None, cur
"service_count": service_count,
"task_count": task_count,
"overdue_count": overdue_count,
"href": f"/clients/{client.id}/edit",
"href": f"/clients/{client.id}",
})
return out
@@ -258,7 +294,14 @@ def _staff_rows(db: Session, tenant_id: int | None, branch_id: int | None, tasks
return rows[:25]
def _billing_rows(db: Session, tenant_id: int | None, branch_id: int | None, fy: str | None) -> dict[str, Any]:
def _billing_rows(
db: Session,
tenant_id: int | None,
branch_id: int | None,
fy: str | None,
current_user,
roles: set[str],
) -> dict[str, Any]:
if BillingInvoice is None or not tenant_id:
return {"available": False, "invoices": [], "invoice_count": 0, "draft_count": 0, "outstanding": Decimal("0")}
stmt = select(BillingInvoice).where(BillingInvoice.tenant_id == tenant_id)
@@ -266,6 +309,15 @@ def _billing_rows(db: Session, tenant_id: int | None, branch_id: int | None, fy:
stmt = stmt.where(BillingInvoice.branch_id == branch_id)
if fy:
stmt = stmt.where(BillingInvoice.financial_year == fy)
if _is_scoped_partner(roles):
allowed_client_ids = select(Client.id).where(
Client.tenant_id == tenant_id,
_partner_client_access_expression(
partner_user_id=int(current_user.id),
tenant_id=tenant_id,
),
)
stmt = stmt.where(BillingInvoice.client_id.in_(allowed_client_ids))
invoices = list(db.execute(stmt.order_by(BillingInvoice.invoice_date.desc(), BillingInvoice.id.desc()).limit(25)).scalars().all())
outstanding = sum((_money(getattr(inv, "balance_amount", 0)) for inv in invoices), Decimal("0"))
return {
@@ -297,7 +349,7 @@ def build_partner_dashboard_payload(db: Session, request, current_user) -> dict[
clients = _load_clients(db, tenant_id, branch_id, current_user, roles)
staff = _staff_rows(db, tenant_id, branch_id, tasks)
billing = _billing_rows(db, tenant_id, branch_id, fy)
billing = _billing_rows(db, tenant_id, branch_id, fy, current_user, roles)
unified_escalations = list_workflow_escalations(db, tenant_id=tenant_id, branch_id=branch_id, assigned_to_user_id=current_user.id)
subscriptions_count = _count(db, _subscription_scope(select(func.count(ClientServiceSubscription.id)), tenant_id, branch_id, current_user, roles, fy)) if tenant_id else 0