from __future__ import annotations import os from dataclasses import dataclass from typing import Any, Mapping from sqlalchemy.orm import Session from govoplan_core.tenancy.scope import Tenant DEFAULT_SMALL_CELL_THRESHOLD = 5 MIN_SMALL_CELL_THRESHOLD = 2 MAX_SMALL_CELL_THRESHOLD = 100 SMALL_CELL_THRESHOLD_ENV = "GOVOPLAN_CAMPAIGN_REPORT_SMALL_CELL_THRESHOLD" CAMPAIGN_REPORT_POLICY_SETTINGS_KEY = "campaign_report_privacy_policy" SMALL_CELL_THRESHOLD_SETTINGS_KEY = "small_cell_threshold" class CampaignReportPrivacyPolicyError(RuntimeError): pass @dataclass(frozen=True, slots=True) class CampaignReportPrivacyPolicy: small_cell_threshold: int source: str deployment_small_cell_threshold: int tenant_small_cell_threshold: int | None = None def as_dict(self) -> dict[str, Any]: return { "small_cell_threshold": self.small_cell_threshold, "source": self.source, "deployment_small_cell_threshold": self.deployment_small_cell_threshold, "tenant_small_cell_threshold": self.tenant_small_cell_threshold, "deployment_setting": SMALL_CELL_THRESHOLD_ENV, "tenant_setting": ( f"tenant.settings.{CAMPAIGN_REPORT_POLICY_SETTINGS_KEY}." f"{SMALL_CELL_THRESHOLD_SETTINGS_KEY}" ), } def effective_campaign_report_privacy_policy( session: Session, *, tenant_id: str, environ: Mapping[str, str] | None = None, ) -> CampaignReportPrivacyPolicy: env = os.environ if environ is None else environ deployment_raw = env.get(SMALL_CELL_THRESHOLD_ENV) deployment_value = _configured_threshold( deployment_raw, source=SMALL_CELL_THRESHOLD_ENV, default=DEFAULT_SMALL_CELL_THRESHOLD, ) tenant = session.get(Tenant, tenant_id) tenant_raw = _tenant_threshold_value(tenant.settings if tenant is not None else None) if tenant_raw is None: return CampaignReportPrivacyPolicy( small_cell_threshold=deployment_value, source="deployment" if deployment_raw not in (None, "") else "deployment_default", deployment_small_cell_threshold=deployment_value, ) tenant_value = _configured_threshold( tenant_raw, source=( f"tenant.settings.{CAMPAIGN_REPORT_POLICY_SETTINGS_KEY}." f"{SMALL_CELL_THRESHOLD_SETTINGS_KEY}" ), ) effective_value = max(deployment_value, tenant_value) return CampaignReportPrivacyPolicy( small_cell_threshold=effective_value, source="tenant" if tenant_value >= deployment_value else "deployment_floor", deployment_small_cell_threshold=deployment_value, tenant_small_cell_threshold=tenant_value, ) def _tenant_threshold_value(settings: Mapping[str, Any] | None) -> object | None: if not isinstance(settings, Mapping): return None policy = settings.get(CAMPAIGN_REPORT_POLICY_SETTINGS_KEY) if not isinstance(policy, Mapping): return None return policy.get(SMALL_CELL_THRESHOLD_SETTINGS_KEY) def _configured_threshold(value: object, *, source: str, default: int | None = None) -> int: if value is None or (isinstance(value, str) and not value.strip()): if default is not None: return default raise CampaignReportPrivacyPolicyError(f"{source} must be configured as an integer") if isinstance(value, bool): raise CampaignReportPrivacyPolicyError(f"{source} must be an integer, not a boolean") try: parsed = int(value) except (TypeError, ValueError) as exc: raise CampaignReportPrivacyPolicyError(f"{source} must be an integer") from exc if str(parsed) != str(value).strip() and not isinstance(value, int): raise CampaignReportPrivacyPolicyError(f"{source} must be an integer") if parsed < MIN_SMALL_CELL_THRESHOLD or parsed > MAX_SMALL_CELL_THRESHOLD: raise CampaignReportPrivacyPolicyError( f"{source} must be between {MIN_SMALL_CELL_THRESHOLD} and {MAX_SMALL_CELL_THRESHOLD}" ) return parsed