from __future__ import annotations from datetime import datetime from typing import Any, Literal from pydantic import BaseModel, ConfigDict, Field, field_validator from govoplan_core.api.v1.schemas import DeltaDeletedItem 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 PermissionItem(BaseModel): scope: str label: str description: str category: str level: Literal["tenant", "system"] class PermissionCatalogResponse(BaseModel): permissions: list[PermissionItem] class AdminOverviewResponse(BaseModel): active_tenant_id: str active_tenant_name: str tenant_count: int | None = None system_account_count: int | None = None system_group_template_count: int | None = None system_role_template_count: int | None = None user_count: int active_user_count: int group_count: int role_count: int active_api_key_count: int capabilities: list[str] = Field(default_factory=list) class TenantAdminItem(BaseModel): id: str slug: str = Field(min_length=1, max_length=100) name: str = Field(min_length=1, max_length=255) description: str | None = None default_locale: str = Field(default="en", min_length=1, max_length=20) settings: dict[str, Any] = Field(default_factory=dict) allow_custom_groups: bool | None = None allow_custom_roles: bool | None = None allow_api_keys: bool | None = None effective_governance: dict[str, bool] = Field(default_factory=dict) is_active: bool counts: dict[str, int] = Field(default_factory=dict) created_at: datetime updated_at: datetime class TenantListResponse(BaseModel): tenants: list[TenantAdminItem] class TenantOwnerCandidate(BaseModel): account_id: str email: str display_name: str | None = None class TenantOwnerCandidateListResponse(BaseModel): accounts: list[TenantOwnerCandidate] class TenantCreateRequest(BaseModel): model_config = ConfigDict(extra="forbid") slug: str name: str owner_account_id: str | None = None description: str | None = None default_locale: str = "en" settings: dict[str, Any] = Field(default_factory=dict) allow_custom_groups: bool | None = None allow_custom_roles: bool | None = None allow_api_keys: bool | None = None class TenantUpdateRequest(BaseModel): model_config = ConfigDict(extra="forbid") name: str | None = Field(default=None, min_length=1, max_length=255) description: str | None = None default_locale: str | None = Field(default=None, min_length=1, max_length=20) settings: dict[str, Any] | None = None allow_custom_groups: bool | None = None allow_custom_roles: bool | None = None allow_api_keys: bool | None = None is_active: bool | None = None class TenantSettingsItem(BaseModel): id: str slug: str name: str default_locale: str = Field(default="en", min_length=1, max_length=20) available_languages: list[dict[str, Any]] = Field(default_factory=list) system_enabled_language_codes: list[str] = Field(default_factory=list) enabled_language_codes: list[str] = Field(default_factory=list) settings: dict[str, Any] = Field(default_factory=dict) class TenantSettingsUpdateRequest(BaseModel): model_config = ConfigDict(extra="forbid") default_locale: str = Field(min_length=1, max_length=20) enabled_language_codes: list[str] | None = None class RoleSummary(BaseModel): id: str slug: str = Field(min_length=1, max_length=100) name: str = Field(min_length=1, max_length=255) description: str | None = None permissions: list[str] = Field(default_factory=list) effective_permission_count: int = 0 is_builtin: bool = False is_assignable: bool = True user_assignments: int = 0 group_assignments: int = 0 level: Literal["tenant", "system"] = "tenant" system_template_id: str | None = None system_required: bool = False class IdentityAccountLinkItem(BaseModel): id: str account_id: str email: str | None = None display_name: str | None = None is_primary: bool = False source: str = "local" class IdentityAdminItem(BaseModel): id: str display_name: str | None = None external_subject: str | None = None source: str = "local" is_active: bool = True accounts: list[IdentityAccountLinkItem] = Field(default_factory=list) created_at: datetime updated_at: datetime class IdentityListResponse(BaseModel): identities: list[IdentityAdminItem] class IdentityCreateRequest(BaseModel): model_config = ConfigDict(extra="forbid") display_name: str | None = Field(default=None, max_length=255) external_subject: str | None = Field(default=None, max_length=255) source: str = Field(default="local", max_length=50) account_ids: list[str] = Field(default_factory=list) primary_account_id: str | None = None is_active: bool = True class IdentityUpdateRequest(BaseModel): model_config = ConfigDict(extra="forbid") display_name: str | None = Field(default=None, max_length=255) external_subject: str | None = Field(default=None, max_length=255) source: str | None = Field(default=None, max_length=50) account_ids: list[str] | None = None primary_account_id: str | None = None is_active: bool | None = None class OrganizationUnitItem(BaseModel): id: str tenant_id: str parent_id: str | None = None slug: str = Field(min_length=1, max_length=100) name: str = Field(min_length=1, max_length=255) description: str | None = None is_active: bool = True created_at: datetime updated_at: datetime class OrganizationUnitListResponse(BaseModel): organization_units: list[OrganizationUnitItem] class OrganizationUnitCreateRequest(BaseModel): model_config = ConfigDict(extra="forbid") slug: str name: str = Field(min_length=1, max_length=255) parent_id: str | None = None description: str | None = None is_active: bool = True class OrganizationUnitUpdateRequest(BaseModel): model_config = ConfigDict(extra="forbid") slug: str | None = None name: str | None = Field(default=None, min_length=1, max_length=255) parent_id: str | None = None description: str | None = None is_active: bool | None = None class FunctionAdminItem(BaseModel): id: str tenant_id: str organization_unit_id: str slug: str = Field(min_length=1, max_length=100) name: str = Field(min_length=1, max_length=255) description: str | None = None role_ids: list[str] = Field(default_factory=list) delegable: bool = False act_in_place_allowed: bool = False is_active: bool = True created_at: datetime updated_at: datetime class FunctionListResponse(BaseModel): functions: list[FunctionAdminItem] class ExternalFunctionRoleMappingItem(BaseModel): id: str tenant_id: str source_module: str function_id: str role_id: str settings: dict[str, Any] = Field(default_factory=dict) created_at: datetime updated_at: datetime class ExternalFunctionRoleMappingListResponse(BaseModel): mappings: list[ExternalFunctionRoleMappingItem] class ExternalFunctionRoleMappingListDeltaResponse(ExternalFunctionRoleMappingListResponse): deleted: list[DeltaDeletedItem] = Field(default_factory=list) watermark: str | None = None has_more: bool = False full: bool = True class ExternalFunctionRoleMappingCreateRequest(BaseModel): model_config = ConfigDict(extra="forbid") function_id: str role_id: str source_module: str = Field(default="organizations", max_length=50) settings: dict[str, Any] = Field(default_factory=dict) class ExternalFunctionRoleMappingUpdateRequest(BaseModel): model_config = ConfigDict(extra="forbid") role_id: str | None = None settings: dict[str, Any] | None = None class FunctionCreateRequest(BaseModel): model_config = ConfigDict(extra="forbid") organization_unit_id: str slug: str name: str = Field(min_length=1, max_length=255) description: str | None = None role_ids: list[str] = Field(default_factory=list) delegable: bool = False act_in_place_allowed: bool = False is_active: bool = True class FunctionUpdateRequest(BaseModel): model_config = ConfigDict(extra="forbid") organization_unit_id: str | None = None slug: str | None = None name: str | None = Field(default=None, min_length=1, max_length=255) description: str | None = None role_ids: list[str] | None = None delegable: bool | None = None act_in_place_allowed: bool | None = None is_active: bool | None = None class FunctionAssignmentAdminItem(BaseModel): id: str tenant_id: str account_id: str identity_id: str | None = None function_id: str organization_unit_id: str applies_to_subunits: bool = False source: str = "direct" delegated_from_assignment_id: str | None = None acting_for_account_id: str | None = None valid_from: datetime | None = None valid_until: datetime | None = None is_active: bool = True created_at: datetime updated_at: datetime class FunctionAssignmentListResponse(BaseModel): assignments: list[FunctionAssignmentAdminItem] class FunctionAssignmentCreateRequest(BaseModel): model_config = ConfigDict(extra="forbid") account_id: str function_id: str organization_unit_id: str | None = None identity_id: str | None = None applies_to_subunits: bool = False source: str = Field(default="direct", max_length=50) delegated_from_assignment_id: str | None = None acting_for_account_id: str | None = None valid_from: datetime | None = None valid_until: datetime | None = None is_active: bool = True class FunctionAssignmentUpdateRequest(BaseModel): model_config = ConfigDict(extra="forbid") identity_id: str | None = None organization_unit_id: str | None = None applies_to_subunits: bool | None = None source: str | None = Field(default=None, max_length=50) delegated_from_assignment_id: str | None = None acting_for_account_id: str | None = None valid_from: datetime | None = None valid_until: datetime | None = None is_active: bool | None = None class FunctionDelegationAdminItem(BaseModel): id: str tenant_id: str function_assignment_id: str delegator_account_id: str delegate_account_id: str mode: Literal["delegate", "act_in_place"] = "delegate" reason: str | None = None valid_from: datetime | None = None valid_until: datetime | None = None revoked_at: datetime | None = None is_active: bool = True created_at: datetime updated_at: datetime class FunctionDelegationListResponse(BaseModel): delegations: list[FunctionDelegationAdminItem] class FunctionDelegationCreateRequest(BaseModel): model_config = ConfigDict(extra="forbid") function_assignment_id: str delegate_account_id: str mode: Literal["delegate", "act_in_place"] = "delegate" reason: str | None = None valid_from: datetime | None = None valid_until: datetime | None = None is_active: bool = True class FunctionDelegationUpdateRequest(BaseModel): model_config = ConfigDict(extra="forbid") reason: str | None = None valid_from: datetime | None = None valid_until: datetime | None = None is_active: bool | None = None revoked: bool | None = None class GroupSummary(BaseModel): id: str slug: str = Field(min_length=1, max_length=100) name: str = Field(min_length=1, max_length=255) description: str | None = None is_active: bool = True member_count: int = 0 member_ids: list[str] = Field(default_factory=list) roles: list[RoleSummary] = Field(default_factory=list) created_at: datetime updated_at: datetime system_template_id: str | None = None system_required: bool = False class UserAdminItem(BaseModel): id: str account_id: str tenant_id: str email: str = Field(min_length=3, max_length=320) display_name: str | None = Field(default=None, max_length=255) is_active: bool account_is_active: bool password_reset_required: bool = False last_login_at: datetime | None = None groups: list[GroupSummary] = Field(default_factory=list) roles: list[RoleSummary] = Field(default_factory=list) function_assignment_ids: list[str] = Field(default_factory=list) function_delegation_ids: list[str] = Field(default_factory=list) effective_scopes: list[str] = Field(default_factory=list) is_owner: bool = False is_last_active_owner: bool = False created_at: datetime updated_at: datetime class UserListResponse(BaseModel): users: list[UserAdminItem] class UserListDeltaResponse(UserListResponse): deleted: list[DeltaDeletedItem] = Field(default_factory=list) watermark: str | None = None has_more: bool = False full: bool = False class UserCreateRequest(BaseModel): model_config = ConfigDict(extra="forbid") email: str display_name: str | None = None password: str | None = Field(default=None, min_length=10) password_reset_required: bool = True is_active: bool = True group_ids: list[str] = Field(default_factory=list) role_ids: list[str] = Field(default_factory=list) class UserCreateResponse(BaseModel): user: UserAdminItem account_created: bool temporary_password: str | None = None class UserUpdateRequest(BaseModel): model_config = ConfigDict(extra="forbid") display_name: str | None = Field(default=None, max_length=255) is_active: bool | None = None group_ids: list[str] | None = None role_ids: list[str] | None = None class GroupListResponse(BaseModel): groups: list[GroupSummary] class GroupListDeltaResponse(GroupListResponse): deleted: list[DeltaDeletedItem] = Field(default_factory=list) watermark: str | None = None has_more: bool = False full: bool = False class GroupCreateRequest(BaseModel): model_config = ConfigDict(extra="forbid") slug: str name: str description: str | None = None is_active: bool = True member_ids: list[str] = Field(default_factory=list) role_ids: list[str] = Field(default_factory=list) class GroupUpdateRequest(BaseModel): model_config = ConfigDict(extra="forbid") name: str | None = Field(default=None, min_length=1, max_length=255) description: str | None = None is_active: bool | None = None member_ids: list[str] | None = None role_ids: list[str] | None = None class RoleListResponse(BaseModel): roles: list[RoleSummary] class RoleListDeltaResponse(RoleListResponse): deleted: list[DeltaDeletedItem] = Field(default_factory=list) watermark: str | None = None has_more: bool = False full: bool = False class RoleCreateRequest(BaseModel): model_config = ConfigDict(extra="forbid") slug: str name: str = Field(min_length=1, max_length=255) description: str | None = None permissions: list[str] = Field(default_factory=list) class RoleUpdateRequest(BaseModel): model_config = ConfigDict(extra="forbid") name: str description: str | None = None permissions: list[str] = Field(default_factory=list) is_assignable: bool = True class SystemAccountItem(BaseModel): account_id: str email: str display_name: str | None = None is_active: bool memberships: list[dict[str, Any]] = Field(default_factory=list) roles: list[RoleSummary] = Field(default_factory=list) last_login_at: datetime | None = None class SystemAccountListResponse(BaseModel): accounts: list[SystemAccountItem] roles: list[RoleSummary] class SystemAccountListDeltaResponse(SystemAccountListResponse): deleted: list[DeltaDeletedItem] = Field(default_factory=list) watermark: str | None = None has_more: bool = False full: bool = False class SystemAccountUpdateRequest(BaseModel): model_config = ConfigDict(extra="forbid") display_name: str | None = Field(default=None, max_length=255) is_active: bool | None = None role_ids: list[str] | None = None class SystemAccountRolesUpdateRequest(BaseModel): model_config = ConfigDict(extra="forbid") role_ids: list[str] = Field(default_factory=list) class SystemMembershipUpdate(BaseModel): model_config = ConfigDict(extra="forbid") tenant_id: str is_active: bool = True role_ids: list[str] = Field(default_factory=list) group_ids: list[str] = Field(default_factory=list) class SystemAccountMembershipsUpdateRequest(BaseModel): model_config = ConfigDict(extra="forbid") memberships: list[SystemMembershipUpdate] = Field(default_factory=list) class SystemAccountCreateRequest(BaseModel): model_config = ConfigDict(extra="forbid") email: str display_name: str | None = None password: str | None = Field(default=None, min_length=10) password_reset_required: bool = True is_active: bool = True role_ids: list[str] = Field(default_factory=list) memberships: list[SystemMembershipUpdate] = Field(default_factory=list) class SystemAccountCreateResponse(BaseModel): account: SystemAccountItem temporary_password: str | None = None 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] class ConfigurationSafetyFieldItem(BaseModel): key: str label: str owner_module: str scope: Literal["system", "tenant", "user", "group", "campaign"] storage: str ui_managed: bool risk: Literal["low", "medium", "high", "destructive"] secret_handling: Literal["none", "reference_only", "env_only"] = "none" required_scopes: list[str] = Field(default_factory=list) dry_run_required: bool = False validation_required: bool = True policy_explanation_required: bool = False audit_event: str | None = None maintenance_required: bool = False two_person_approval_required: bool = False rollback_history_required: bool = False notes: str | None = None class ConfigurationSafetyCatalogResponse(BaseModel): fields: list[ConfigurationSafetyFieldItem] class ConfigurationChangeSafetyPlanRequest(BaseModel): model_config = ConfigDict(extra="forbid") key: str value: Any = None dry_run: bool = False maintenance_mode: bool = False approval_count: int = Field(default=0, ge=0) class ConfigurationChangeSafetyPlanResponse(BaseModel): plan: dict[str, Any] class ConfigurationChangeRequestCreateRequest(BaseModel): model_config = ConfigDict(extra="forbid") key: str value: Any = None dry_run: bool = False target: dict[str, Any] = Field(default_factory=dict) reason: str | None = Field(default=None, max_length=1000) class ConfigurationChangeRequestApproveRequest(BaseModel): model_config = ConfigDict(extra="forbid") reason: str | None = Field(default=None, max_length=1000) class ConfigurationChangeRequestResponse(BaseModel): request: dict[str, Any] class ConfigurationControlSnapshotResponse(BaseModel): requests: list[dict[str, Any]] = Field(default_factory=list) history: list[dict[str, Any]] = Field(default_factory=list) class ConfigurationControlDeltaResponse(ConfigurationControlSnapshotResponse): deleted: list[DeltaDeletedItem] = Field(default_factory=list) watermark: str | None = None has_more: bool = False full: bool = False class ConfigurationPackageCatalogResponse(BaseModel): validation: dict[str, Any] class ConfigurationPackageRunRequest(BaseModel): model_config = ConfigDict(extra="forbid") package: dict[str, Any] tenant_id: str | None = None supplied_data: dict[str, Any] = Field(default_factory=dict) change_request_id: str | None = None class ConfigurationPackageDryRunResponse(BaseModel): diagnostics: list[dict[str, Any]] = Field(default_factory=list) required_data: list[dict[str, Any]] = Field(default_factory=list) plan: list[dict[str, Any]] = Field(default_factory=list) class ConfigurationPackageApplyResponse(BaseModel): diagnostics: list[dict[str, Any]] = Field(default_factory=list) created_refs: dict[str, str] = Field(default_factory=dict) updated_refs: dict[str, str] = Field(default_factory=dict) class ConfigurationPackageExportRequest(BaseModel): model_config = ConfigDict(extra="forbid") tenant_id: str | None = None scopes: list[str] = Field(default_factory=list) module_ids: list[str] = Field(default_factory=list) object_refs: list[str] = Field(default_factory=list) class ConfigurationPackageExportResponse(BaseModel): fragments: list[dict[str, Any]] = Field(default_factory=list) data_requirements: list[dict[str, Any]] = Field(default_factory=list) diagnostics: list[dict[str, Any]] = Field(default_factory=list) class SystemSettingsItem(BaseModel): default_locale: str = "en" allow_tenant_custom_groups: bool = True allow_tenant_custom_roles: bool = True allow_tenant_api_keys: bool = True privacy_retention_policy: PrivacyRetentionPolicyItem = Field(default_factory=PrivacyRetentionPolicyItem) available_languages: list[dict[str, Any]] = Field(default_factory=list) enabled_language_codes: list[str] = Field(default_factory=list) settings: dict[str, Any] = Field(default_factory=dict) class SystemSettingsUpdateRequest(BaseModel): model_config = ConfigDict(extra="forbid") default_locale: str = Field(min_length=1, max_length=20) allow_tenant_custom_groups: bool allow_tenant_custom_roles: bool allow_tenant_api_keys: bool privacy_retention_policy: PrivacyRetentionPolicyItem | None = None available_languages: list[dict[str, Any]] | None = None enabled_language_codes: list[str] | None = None class ApiKeyAdminItem(BaseModel): id: str user_id: str user_email: str name: str prefix: str scopes: list[str] = Field(default_factory=list) expires_at: datetime | None = None last_used_at: datetime | None = None revoked_at: datetime | None = None created_at: datetime class ApiKeyListResponse(BaseModel): api_keys: list[ApiKeyAdminItem] class ApiKeyListDeltaResponse(ApiKeyListResponse): deleted: list[DeltaDeletedItem] = Field(default_factory=list) watermark: str | None = None has_more: bool = False full: bool = False class AdminApiKeyCreateRequest(BaseModel): model_config = ConfigDict(extra="forbid") name: str = Field(min_length=1, max_length=255) user_id: str | None = None scopes: list[str] = Field(default_factory=list) expires_at: datetime | None = None class AdminApiKeyCreateResponse(ApiKeyAdminItem): secret: str class AuditAdminItem(BaseModel): id: str scope: Literal["tenant", "system"] = "tenant" tenant_id: str | None = None actor_email: str | None = None action: str object_type: str | None = None object_id: str | None = None details: dict[str, Any] = Field(default_factory=dict) created_at: datetime class AuditAdminListResponse(BaseModel): items: list[AuditAdminItem] total: int page: int = 1 page_size: int = 100 pages: int = 1 cursor: str | None = None next_cursor: str | None = None