add-ideas/govoplan-core
1
0
Fork 0
Code Issues 45 Pull Requests Actions Packages Projects Releases Wiki Activity

Prepare v0.1.0 release dependencies

Browse Source
This commit is contained in:
Albrecht Degering
2026-06-24 20:02:49 +02:00
parent ce30b4d054
commit 9c2b84a64a
27 changed files with 900 additions and 48 deletions
Download Patch File Download Diff File Expand all files Collapse all files

26
src/govoplan_core/api/v1/admin.py
View File

@@ -117,8 +117,10 @@ from govoplan_core.privacy.retention import (
PrivacyPolicyError,
apply_retention_policy,
effective_privacy_policy,
effective_privacy_policy_sources,
get_privacy_policy_for_scope,
parent_privacy_policy,
parent_privacy_policy_sources,
privacy_policy_from_settings,
set_privacy_policy,
set_privacy_policy_for_scope,
@@ -1235,6 +1237,8 @@ def read_privacy_retention_policy(
policy=policy,
effective_policy=PrivacyRetentionPolicyItem.model_validate(effective.model_dump(mode="json")),
parent_policy=PrivacyRetentionPolicyItem.model_validate(parent.model_dump(mode="json")) if parent else None,
effective_policy_sources=_effective_privacy_policy_sources_for_response(session, tenant_id=principal.tenant_id, scope_type=clean_scope, scope_id=scope_id),
parent_policy_sources=_parent_privacy_policy_sources_for_response(session, tenant_id=principal.tenant_id, scope_type=clean_scope, scope_id=scope_id),
)
except PrivacyPolicyError as exc:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc)) from exc
@@ -1267,12 +1271,34 @@ def write_privacy_retention_policy(
policy=policy,
effective_policy=PrivacyRetentionPolicyItem.model_validate(effective.model_dump(mode="json")),
parent_policy=PrivacyRetentionPolicyItem.model_validate(parent.model_dump(mode="json")) if parent else None,
effective_policy_sources=_effective_privacy_policy_sources_for_response(session, tenant_id=principal.tenant_id, scope_type=clean_scope, scope_id=scope_id),
parent_policy_sources=_parent_privacy_policy_sources_for_response(session, tenant_id=principal.tenant_id, scope_type=clean_scope, scope_id=scope_id),
)
except PrivacyPolicyError as exc:
session.rollback()
raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail=str(exc)) from exc
def _parent_privacy_policy_sources_for_response(session: Session, *, tenant_id: str, scope_type: str, scope_id: str | None):
if scope_type == "system":
return []
return parent_privacy_policy_sources(session, tenant_id=tenant_id, scope_type=scope_type, scope_id=scope_id or (tenant_id if scope_type == "tenant" else None))
def _effective_privacy_policy_sources_for_response(session: Session, *, tenant_id: str, scope_type: str, scope_id: str | None):
if scope_type == "system":
return effective_privacy_policy_sources(session)
if scope_type == "tenant":
return effective_privacy_policy_sources(session, tenant_id=scope_id or tenant_id)
if scope_type == "campaign" and scope_id:
return effective_privacy_policy_sources(session, campaign_id=scope_id)
if scope_type == "user" and scope_id:
return effective_privacy_policy_sources(session, tenant_id=tenant_id, owner_user_id=scope_id)
if scope_type == "group" and scope_id:
return effective_privacy_policy_sources(session, tenant_id=tenant_id, owner_group_id=scope_id)
return effective_privacy_policy_sources(session, tenant_id=tenant_id)
def _parent_privacy_policy_for_response(session: Session, *, tenant_id: str, scope_type: str, scope_id: str | None):
if scope_type == "system":
return None

9
src/govoplan_core/api/v1/admin_schemas.py
View File

@@ -312,6 +312,13 @@ class SystemAccountCreateResponse(BaseModel):
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)
class PrivacyRetentionPolicyItem(BaseModel):
model_config = ConfigDict(extra="forbid")
@@ -360,6 +367,8 @@ class PrivacyRetentionPolicyScopeResponse(BaseModel):
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):

87
src/govoplan_core/privacy/retention.py
View File

@@ -304,6 +304,93 @@ def parent_privacy_policy(session: Session, *, tenant_id: str, scope_type: str,
return policy
def _retention_policy_source_fields(patch: dict[str, Any], *, baseline: bool = False) -> list[str]:
fields: list[str] = []
for key in RETENTION_POLICY_FIELD_KEYS:
if key in patch and patch.get(key) is not None:
fields.append(key)
if isinstance(patch.get("allow_lower_level_limits"), dict) and patch["allow_lower_level_limits"]:
fields.append("allow_lower_level_limits")
if baseline and not fields:
fields.append("defaults")
return fields
def _retention_policy_source_step(scope_type: str, label: str, scope_id: str | None, patch: dict[str, Any], *, baseline: bool = False) -> dict[str, Any]:
return {
"scope_type": scope_type,
"scope_id": scope_id,
"label": label,
"applied_fields": _retention_policy_source_fields(patch, baseline=baseline),
}
def effective_privacy_policy_sources(
session: Session,
*,
tenant_id: str | None = None,
owner_user_id: str | None = None,
owner_group_id: str | None = None,
campaign_id: str | None = None,
) -> list[dict[str, Any]]:
system_settings = get_system_settings(session)
sources = [_retention_policy_source_step("system", "System", None, _privacy_policy_patch_from_settings(system_settings.settings or {}), baseline=True)]
campaign: Campaign | None = None
if campaign_id:
campaign = session.get(Campaign, campaign_id)
if campaign is None:
raise PrivacyPolicyError("Campaign not found for privacy policy")
tenant_id = campaign.tenant_id
owner_user_id = campaign.owner_user_id
owner_group_id = campaign.owner_group_id
if tenant_id:
tenant = session.get(Tenant, tenant_id)
if tenant is None:
raise PrivacyPolicyError("Tenant not found for privacy policy")
sources.append(_retention_policy_source_step("tenant", "Tenant", tenant.id, _privacy_policy_patch_from_settings(tenant.settings or {})))
if owner_user_id:
user = session.get(User, owner_user_id)
if user and (not tenant_id or user.tenant_id == tenant_id):
sources.append(_retention_policy_source_step("user", "Owner user", user.id, _privacy_policy_patch_from_settings(user.settings or {})))
if owner_group_id:
group = session.get(Group, owner_group_id)
if group and (not tenant_id or group.tenant_id == tenant_id):
sources.append(_retention_policy_source_step("group", "Owner group", group.id, _privacy_policy_patch_from_settings(group.settings or {})))
if campaign is not None:
sources.append(_retention_policy_source_step("campaign", "Campaign", campaign.id, _privacy_policy_patch_from_settings(campaign.settings or {})))
return sources
def parent_privacy_policy_sources(session: Session, *, tenant_id: str, scope_type: str, scope_id: str | None = None) -> list[dict[str, Any]]:
clean_scope = scope_type.strip().casefold()
if clean_scope == "system":
return []
system_settings = get_system_settings(session)
sources = [_retention_policy_source_step("system", "System", None, _privacy_policy_patch_from_settings(system_settings.settings or {}), baseline=True)]
if clean_scope == "tenant":
return sources
tenant = session.get(Tenant, tenant_id)
if tenant is None:
raise PrivacyPolicyError("Tenant not found for privacy policy")
sources.append(_retention_policy_source_step("tenant", "Tenant", tenant.id, _privacy_policy_patch_from_settings(tenant.settings or {})))
if clean_scope in {"user", "group"}:
return sources
if clean_scope != "campaign" or not scope_id:
return sources
campaign = session.get(Campaign, scope_id)
if campaign is None or campaign.tenant_id != tenant_id:
raise PrivacyPolicyError("Campaign not found for privacy policy")
if campaign.owner_user_id:
user = session.get(User, campaign.owner_user_id)
if user and user.tenant_id == tenant_id:
sources.append(_retention_policy_source_step("user", "Owner user", user.id, _privacy_policy_patch_from_settings(user.settings or {})))
if campaign.owner_group_id:
group = session.get(Group, campaign.owner_group_id)
if group and group.tenant_id == tenant_id:
sources.append(_retention_policy_source_step("group", "Owner group", group.id, _privacy_policy_patch_from_settings(group.settings or {})))
return sources
def get_privacy_policy_for_scope(session: Session, *, tenant_id: str, scope_type: str, scope_id: str | None = None) -> dict[str, Any]:
clean_scope = scope_type.strip().casefold()
if clean_scope == "system":