106 lines
3.8 KiB
Python
106 lines
3.8 KiB
Python
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
|
|
from sqlalchemy.orm import Session
|
|
|
|
from govoplan_core.admin.common import AdminValidationError
|
|
from govoplan_core.admin.settings import get_system_settings
|
|
from govoplan_core.core.access import CAPABILITY_ACCESS_ADMINISTRATION, AccessAdministration
|
|
from govoplan_core.core.runtime import get_registry
|
|
from govoplan_core.tenancy.scope import Tenant
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class EffectiveTenantGovernance:
|
|
allow_custom_groups: bool
|
|
allow_custom_roles: bool
|
|
allow_api_keys: bool
|
|
|
|
|
|
def _narrowing_bool(system_allows: bool, tenant_override: bool | None) -> bool:
|
|
if not system_allows:
|
|
return False
|
|
return tenant_override is not False
|
|
|
|
|
|
def effective_tenant_governance(session: Session, tenant: Tenant) -> EffectiveTenantGovernance:
|
|
defaults = get_system_settings(session)
|
|
return EffectiveTenantGovernance(
|
|
allow_custom_groups=_narrowing_bool(defaults.allow_tenant_custom_groups, tenant.allow_custom_groups),
|
|
allow_custom_roles=_narrowing_bool(defaults.allow_tenant_custom_roles, tenant.allow_custom_roles),
|
|
allow_api_keys=_narrowing_bool(defaults.allow_tenant_api_keys, tenant.allow_api_keys),
|
|
)
|
|
|
|
|
|
def assert_tenant_governance_override_allowed(session: Session, *, field: str, value: bool | None) -> None:
|
|
if value is not True:
|
|
return
|
|
defaults = get_system_settings(session)
|
|
blocked = {
|
|
"allow_custom_groups": not defaults.allow_tenant_custom_groups,
|
|
"allow_custom_roles": not defaults.allow_tenant_custom_roles,
|
|
"allow_api_keys": not defaults.allow_tenant_api_keys,
|
|
}
|
|
if blocked.get(field):
|
|
raise AdminValidationError("Tenant governance cannot explicitly allow a capability denied by system settings.")
|
|
|
|
|
|
def _tenant_module_counts(
|
|
session: Session,
|
|
tenant_id: str,
|
|
*,
|
|
module_ids: tuple[str, ...] | None = None,
|
|
) -> dict[str, int]:
|
|
registry = get_registry()
|
|
if registry is None or not hasattr(registry, "tenant_summary_providers"):
|
|
return {}
|
|
counts: dict[str, int] = {}
|
|
providers = registry.tenant_summary_providers()
|
|
if module_ids is not None:
|
|
providers = {
|
|
module_id: provider
|
|
for module_id, provider in providers.items()
|
|
if module_id in module_ids
|
|
}
|
|
for provider in providers.values():
|
|
provided = provider(session, tenant_id)
|
|
counts.update({str(key): int(value) for key, value in provided.items()})
|
|
return counts
|
|
|
|
|
|
def _access_administration() -> AccessAdministration | None:
|
|
registry = get_registry()
|
|
if registry is None or not hasattr(registry, "has_capability") or not registry.has_capability(CAPABILITY_ACCESS_ADMINISTRATION):
|
|
return None
|
|
capability = registry.require_capability(CAPABILITY_ACCESS_ADMINISTRATION)
|
|
if not isinstance(capability, AccessAdministration):
|
|
raise TypeError("Access administration capability is invalid")
|
|
return capability
|
|
|
|
|
|
def tenant_counts(
|
|
session: Session,
|
|
tenant_id: str,
|
|
*,
|
|
module_ids: tuple[str, ...] | None = None,
|
|
) -> dict[str, int]:
|
|
module_counts = _tenant_module_counts(
|
|
session,
|
|
tenant_id,
|
|
module_ids=module_ids,
|
|
)
|
|
access_administration = _access_administration()
|
|
access_counts = access_administration.tenant_counts(session, tenant_id) if access_administration is not None else {}
|
|
|
|
return {
|
|
**module_counts,
|
|
"users": int(access_counts.get("users", 0)),
|
|
"active_users": int(access_counts.get("active_users", 0)),
|
|
"groups": int(access_counts.get("groups", 0)),
|
|
"campaigns": int(module_counts.get("campaigns", 0)),
|
|
"files": int(module_counts.get("files", 0)),
|
|
"api_keys": int(access_counts.get("api_keys", 0)),
|
|
"active_api_keys": int(access_counts.get("active_api_keys", access_counts.get("api_keys", 0))),
|
|
}
|