110 lines
4.0 KiB
Python
110 lines
4.0 KiB
Python
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
|