Release v0.1.5

src/govoplan_policy/backend/api/v1/schemas.py

@@ -0,0 +1,110 @@
+from __future__ import annotations
+
+from datetime import datetime
+from typing import Any, Literal
+
+from pydantic import BaseModel, ConfigDict, Field, field_validator
+
+
+RETENTION_DAY_KEYS = (
+    "raw_campaign_json_retention_days",
+    "generated_eml_retention_days",
+    "stored_report_detail_retention_days",
+    "mock_mailbox_retention_days",
+    "audit_detail_retention_days",
+)
+
+
+RETENTION_POLICY_FIELD_KEYS = (
+    "store_raw_campaign_json",
+    *RETENTION_DAY_KEYS,
+    "audit_detail_level",
+)
+
+
+def default_allow_lower_level_limits() -> dict[str, bool]:
+    return {key: True for key in RETENTION_POLICY_FIELD_KEYS}
+
+
+def normalize_allow_lower_level_limits(value: Any, *, fill_defaults: bool) -> dict[str, bool] | None:
+    if value in (None, ""):
+        return default_allow_lower_level_limits() if fill_defaults else None
+    if not isinstance(value, dict):
+        raise ValueError("allow_lower_level_limits must be an object")
+    normalized = default_allow_lower_level_limits() if fill_defaults else {}
+    for key, allowed in value.items():
+        clean_key = str(key)
+        if clean_key not in RETENTION_POLICY_FIELD_KEYS:
+            raise ValueError(f"Unknown retention policy field: {clean_key}")
+        normalized[clean_key] = bool(allowed)
+    return normalized
+
+
+class PolicySourceStepItem(BaseModel):
+    scope_type: str
+    scope_id: str | None = None
+    label: str
+    applied_fields: list[str] = Field(default_factory=list)
+    policy: dict[str, Any] = Field(default_factory=dict)
+
+
+class PrivacyRetentionPolicyItem(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+
+    store_raw_campaign_json: bool = True
+    raw_campaign_json_retention_days: int | None = Field(default=None, ge=0)
+    generated_eml_retention_days: int | None = Field(default=None, ge=0)
+    stored_report_detail_retention_days: int | None = Field(default=None, ge=0)
+    mock_mailbox_retention_days: int | None = Field(default=None, ge=0)
+    audit_detail_retention_days: int | None = Field(default=None, ge=0)
+    audit_detail_level: Literal["full", "redacted", "minimal"] = "full"
+    allow_lower_level_limits: dict[str, bool] = Field(default_factory=default_allow_lower_level_limits)
+
+    @field_validator("allow_lower_level_limits", mode="before")
+    @classmethod
+    def _normalize_allow_lower_level_limits(cls, value: Any) -> Any:
+        return normalize_allow_lower_level_limits(value, fill_defaults=True)
+
+
+class PrivacyRetentionPolicyPatchItem(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+
+    store_raw_campaign_json: bool | None = None
+    raw_campaign_json_retention_days: int | None = Field(default=None, ge=0)
+    generated_eml_retention_days: int | None = Field(default=None, ge=0)
+    stored_report_detail_retention_days: int | None = Field(default=None, ge=0)
+    mock_mailbox_retention_days: int | None = Field(default=None, ge=0)
+    audit_detail_retention_days: int | None = Field(default=None, ge=0)
+    audit_detail_level: Literal["full", "redacted", "minimal"] | None = None
+    allow_lower_level_limits: dict[str, bool] | None = None
+
+    @field_validator("allow_lower_level_limits", mode="before")
+    @classmethod
+    def _normalize_allow_lower_level_limits(cls, value: Any) -> Any:
+        return normalize_allow_lower_level_limits(value, fill_defaults=False)
+
+
+class PrivacyRetentionPolicyScopeRequest(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+
+    policy: PrivacyRetentionPolicyPatchItem = Field(default_factory=PrivacyRetentionPolicyPatchItem)
+
+
+class PrivacyRetentionPolicyScopeResponse(BaseModel):
+    scope_type: Literal["system", "tenant", "user", "group", "campaign"]
+    scope_id: str | None = None
+    policy: dict[str, Any]
+    effective_policy: PrivacyRetentionPolicyItem
+    parent_policy: PrivacyRetentionPolicyItem | None = None
+    effective_policy_sources: list[PolicySourceStepItem] = Field(default_factory=list)
+    parent_policy_sources: list[PolicySourceStepItem] = Field(default_factory=list)
+
+
+class RetentionRunRequest(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+
+    dry_run: bool = True
+
+
+class RetentionRunResponse(BaseModel):
+    result: dict[str, Any]