221 lines
7.5 KiB
Python
221 lines
7.5 KiB
Python
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from collections.abc import Mapping, Sequence
|
|
from typing import Protocol, runtime_checkable
|
|
|
|
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
|
|
|
|
|
|
@runtime_checkable
|
|
class _AccessTenantCountsBatchProvider(Protocol):
|
|
def tenant_counts_many(
|
|
self,
|
|
session: object,
|
|
tenant_ids: Sequence[str],
|
|
) -> Mapping[str, Mapping[str, int]]:
|
|
...
|
|
|
|
|
|
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))),
|
|
}
|
|
|
|
|
|
def tenant_counts_many(
|
|
session: Session,
|
|
tenant_ids: Sequence[str],
|
|
*,
|
|
module_ids: tuple[str, ...] | None = None,
|
|
) -> dict[str, dict[str, int]]:
|
|
"""Collect list-page summaries while preserving legacy provider support."""
|
|
|
|
normalized_ids = tuple(
|
|
dict.fromkeys(
|
|
str(tenant_id).strip()
|
|
for tenant_id in tenant_ids
|
|
if str(tenant_id).strip()
|
|
)
|
|
)
|
|
counts_by_tenant: dict[str, dict[str, int]] = {
|
|
tenant_id: {} for tenant_id in normalized_ids
|
|
}
|
|
if not normalized_ids:
|
|
return counts_by_tenant
|
|
|
|
registry = get_registry()
|
|
if registry is not None and hasattr(registry, "tenant_summary_providers"):
|
|
providers = dict(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
|
|
}
|
|
batch_providers = (
|
|
dict(registry.tenant_summary_batch_providers())
|
|
if hasattr(registry, "tenant_summary_batch_providers")
|
|
else {}
|
|
)
|
|
for module_id, provider in providers.items():
|
|
batch_provider = batch_providers.get(module_id)
|
|
if batch_provider is None:
|
|
provided_by_tenant = {
|
|
tenant_id: provider(session, tenant_id)
|
|
for tenant_id in normalized_ids
|
|
}
|
|
else:
|
|
provided_by_tenant = batch_provider(session, normalized_ids)
|
|
_merge_tenant_counts(
|
|
counts_by_tenant,
|
|
provided_by_tenant,
|
|
tenant_ids=normalized_ids,
|
|
)
|
|
|
|
access_administration = _access_administration()
|
|
if isinstance(access_administration, _AccessTenantCountsBatchProvider):
|
|
access_counts = access_administration.tenant_counts_many(
|
|
session,
|
|
normalized_ids,
|
|
)
|
|
elif access_administration is not None:
|
|
access_counts = {
|
|
tenant_id: access_administration.tenant_counts(session, tenant_id)
|
|
for tenant_id in normalized_ids
|
|
}
|
|
else:
|
|
access_counts = {}
|
|
_merge_tenant_counts(
|
|
counts_by_tenant,
|
|
access_counts,
|
|
tenant_ids=normalized_ids,
|
|
)
|
|
|
|
return {
|
|
tenant_id: _normalized_tenant_counts(counts_by_tenant[tenant_id])
|
|
for tenant_id in normalized_ids
|
|
}
|
|
|
|
|
|
def _merge_tenant_counts(
|
|
target: dict[str, dict[str, int]],
|
|
provided_by_tenant: Mapping[str, Mapping[str, int]],
|
|
*,
|
|
tenant_ids: Sequence[str],
|
|
) -> None:
|
|
for tenant_id in tenant_ids:
|
|
provided = provided_by_tenant.get(tenant_id, {})
|
|
target[tenant_id].update(
|
|
{str(key): int(value) for key, value in provided.items()}
|
|
)
|
|
|
|
|
|
def _normalized_tenant_counts(counts: Mapping[str, int]) -> dict[str, int]:
|
|
return {
|
|
**{str(key): int(value) for key, value in counts.items()},
|
|
"users": int(counts.get("users", 0)),
|
|
"active_users": int(counts.get("active_users", 0)),
|
|
"groups": int(counts.get("groups", 0)),
|
|
"campaigns": int(counts.get("campaigns", 0)),
|
|
"files": int(counts.get("files", 0)),
|
|
"api_keys": int(counts.get("api_keys", 0)),
|
|
"active_api_keys": int(
|
|
counts.get("active_api_keys", counts.get("api_keys", 0))
|
|
),
|
|
}
|