110 lines
3.5 KiB
Python
110 lines
3.5 KiB
Python
"""
|
|
P0 3D Capability Matrix — Capability Resolver Service
|
|
|
|
Resolves the effective feature capabilities for a given organization by merging:
|
|
1. The SubscriptionTier's feature_capabilities (tier defaults)
|
|
2. The Organization's custom_features (org-specific overrides)
|
|
|
|
Superadmin "Fast-Pass": If the user's role is SUPERADMIN, all capabilities
|
|
are bypassed and a special {"_is_superadmin": True} marker is returned.
|
|
"""
|
|
|
|
import logging
|
|
from typing import Dict, Any, Optional
|
|
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from sqlalchemy.orm import joinedload
|
|
|
|
from app.models.identity import User, UserRole
|
|
from app.models.marketplace.organization import Organization
|
|
from app.models.core_logic import SubscriptionTier
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
async def resolve_org_capabilities(
|
|
user: User,
|
|
org_id: Optional[int],
|
|
db: AsyncSession,
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Resolve the effective 3D capability matrix for a user within an organization.
|
|
|
|
Business Logic:
|
|
1. Fast-Pass: If user.role is SUPERADMIN, return {"_is_superadmin": True}.
|
|
2. If no org_id is provided, return an empty dict (no capabilities).
|
|
3. Fetch the Organization by org_id, joined with its SubscriptionTier.
|
|
4. tier_features = tier.feature_capabilities (or {} if None/missing).
|
|
5. org_features = org.custom_features (or {} if None/missing).
|
|
6. Merge: {**tier_features, **org_features} (org overrides tier).
|
|
7. Return the merged dict.
|
|
|
|
Args:
|
|
user: The authenticated User instance.
|
|
org_id: The active organization ID (may be None).
|
|
db: Async SQLAlchemy session.
|
|
|
|
Returns:
|
|
A dict of resolved capability flags.
|
|
"""
|
|
# ── Fast-Pass: SUPERADMIN bypasses everything ──
|
|
if user.role == UserRole.SUPERADMIN:
|
|
logger.debug(
|
|
"Capability Fast-Pass for user=%s (SUPERADMIN)",
|
|
user.id,
|
|
)
|
|
return {"_is_superadmin": True}
|
|
|
|
# ── No active organization → no capabilities ──
|
|
if org_id is None:
|
|
logger.debug(
|
|
"No active org for user=%s, returning empty capabilities",
|
|
user.id,
|
|
)
|
|
return {}
|
|
|
|
# ── Fetch Organization + SubscriptionTier ──
|
|
stmt = (
|
|
select(Organization)
|
|
.options(
|
|
joinedload(Organization.subscription_tier)
|
|
)
|
|
.where(Organization.id == org_id)
|
|
)
|
|
result = await db.execute(stmt)
|
|
org = result.unique().scalar_one_or_none()
|
|
|
|
if not org:
|
|
logger.warning(
|
|
"Organization id=%s not found for user=%s, returning empty capabilities",
|
|
org_id,
|
|
user.id,
|
|
)
|
|
return {}
|
|
|
|
# ── Extract tier features ──
|
|
tier_features: Dict[str, Any] = {}
|
|
if org.subscription_tier_id is not None:
|
|
# The subscription_tier is loaded via joinedload
|
|
tier = org.subscription_tier # type: Optional[SubscriptionTier]
|
|
if tier and tier.feature_capabilities:
|
|
tier_features = dict(tier.feature_capabilities)
|
|
|
|
# ── Extract org custom features ──
|
|
org_features: Dict[str, Any] = {}
|
|
if org.custom_features:
|
|
org_features = dict(org.custom_features)
|
|
|
|
# ── Merge: org custom_features override tier defaults ──
|
|
merged = {**tier_features, **org_features}
|
|
|
|
logger.debug(
|
|
"Resolved capabilities for org=%s user=%s: %s",
|
|
org_id,
|
|
user.id,
|
|
merged,
|
|
)
|
|
|
|
return merged
|