97 lines
3.2 KiB
Python
97 lines
3.2 KiB
Python
from __future__ import annotations
|
|
|
|
import uuid
|
|
from typing import Any, Dict, Optional, Set
|
|
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.models.auth.access_model import Access
|
|
from app.models.auth.module_access_model import ModuleAccess
|
|
from app.models.auth.plan_access_model import PlanAccess
|
|
from app.models.auth.plan_module_access_model import PlanModuleAccess
|
|
from app.models.auth.tenant_model import Tenant
|
|
from app.models.auth.user_model import User
|
|
|
|
|
|
class SubscriptionEntitlementService:
|
|
@staticmethod
|
|
def get_tenant(db: Session, tenant_id: Optional[uuid.UUID]) -> Optional[Tenant]:
|
|
if not tenant_id:
|
|
return None
|
|
return db.query(Tenant).filter(Tenant.id == tenant_id).first()
|
|
|
|
@staticmethod
|
|
def get_plan_access_codes(
|
|
db: Session,
|
|
tenant_id: Optional[uuid.UUID],
|
|
) -> Set[str]:
|
|
tenant = SubscriptionEntitlementService.get_tenant(db, tenant_id)
|
|
if not tenant or not tenant.plan_id:
|
|
return set()
|
|
|
|
rows = (
|
|
db.query(Access.access_code)
|
|
.join(PlanAccess, PlanAccess.access_id == Access.id)
|
|
.filter(PlanAccess.plan_id == tenant.plan_id)
|
|
.all()
|
|
)
|
|
return {row[0] for row in rows}
|
|
|
|
@staticmethod
|
|
def get_plan_module_access_codes(
|
|
db: Session,
|
|
tenant_id: Optional[uuid.UUID],
|
|
module_id: Optional[uuid.UUID] = None,
|
|
) -> Set[str]:
|
|
tenant = SubscriptionEntitlementService.get_tenant(db, tenant_id)
|
|
if not tenant or not tenant.plan_id:
|
|
return set()
|
|
|
|
query = (
|
|
db.query(ModuleAccess.access_code)
|
|
.join(
|
|
PlanModuleAccess,
|
|
PlanModuleAccess.module_access_id == ModuleAccess.id,
|
|
)
|
|
.filter(PlanModuleAccess.plan_id == tenant.plan_id)
|
|
)
|
|
if module_id:
|
|
query = query.filter(ModuleAccess.module_id == module_id)
|
|
|
|
rows = query.all()
|
|
return {row[0] for row in rows}
|
|
|
|
@staticmethod
|
|
def get_effective_access_codes(db: Session, user: User) -> Set[str]:
|
|
role_codes = set()
|
|
if user.role:
|
|
role_codes = {
|
|
ra.access.access_code
|
|
for ra in user.role.role_accesses
|
|
if ra.access is not None
|
|
}
|
|
|
|
plan_codes = SubscriptionEntitlementService.get_plan_access_codes(
|
|
db, user.tenant_id
|
|
)
|
|
return role_codes | plan_codes
|
|
|
|
@staticmethod
|
|
def get_subscription_summary(
|
|
db: Session, tenant_id: Optional[uuid.UUID]
|
|
) -> Optional[Dict[str, Any]]:
|
|
tenant = SubscriptionEntitlementService.get_tenant(db, tenant_id)
|
|
if not tenant:
|
|
return None
|
|
|
|
plan_name = tenant.plan.name if tenant.plan else None
|
|
return {
|
|
"plan_id": str(tenant.plan_id) if tenant.plan_id else None,
|
|
"plan_name": plan_name,
|
|
"max_users_allowed": tenant.plan.max_users_allowed if tenant.plan else None,
|
|
"start_date": tenant.start_date.isoformat() if tenant.start_date else None,
|
|
"end_date": tenant.end_date.isoformat() if tenant.end_date else None,
|
|
"status": tenant.status,
|
|
"is_active": tenant.is_active,
|
|
}
|