GovOPlaN/govoplan-campaign
1
0
Fork 0
Code Issues 47 Pull Requests Actions Packages Projects Releases Wiki Activity

intermittent commit

Browse Source
This commit is contained in:
Albrecht Degering
2026-07-14 13:22:10 +02:00
parent 3f0b14a726
commit 2e593b7fa4
31 changed files with 3523 additions and 1672 deletions
Download Patch File Download Diff File Expand all files Collapse all files

56
src/govoplan_campaign/backend/change_tracking.py
View File

@@ -3,10 +3,17 @@ from __future__ import annotations
from datetime import datetime
from typing import Any
from sqlalchemy import event, inspect
from sqlalchemy import event
from sqlalchemy.orm import Session as OrmSession
from govoplan_core.core.change_sequence import record_change
from govoplan_core.core.sqlalchemy_change_tracking import (
ensure_object_id,
has_attr_changes,
object_state,
operation_for_object,
previous_value,
)
from govoplan_campaign.backend.db.models import (
Campaign,
CampaignIssue,
@@ -85,10 +92,10 @@ def _record_campaign_change(session: OrmSession, campaign: Campaign) -> None:
def _record_share_visibility_change(session: OrmSession, share: CampaignShare) -> None:
state = inspect(share)
state = object_state(share)
if not share.campaign_id:
return
if not state.deleted and not state.pending and not _has_attr_changes(
if not state.deleted and not state.pending and not has_attr_changes(
state,
("campaign_id", "target_type", "target_id", "permission", "revoked_at"),
):
@@ -271,12 +278,12 @@ def _record_attempt_change(session: OrmSession, attempt: SendAttempt | ImapAppen
def _operation_for_campaign(obj: Campaign, *, changed_attrs: tuple[str, ...]) -> str | None:
state = inspect(obj)
state = object_state(obj)
if state.pending:
return "created"
if state.deleted:
return "deleted"
if not _has_attr_changes(state, changed_attrs):
if not has_attr_changes(state, changed_attrs):
return None
status_history = state.attrs.status.history
if status_history.has_changes() and any(value == "deleted" for value in status_history.added):
@@ -285,38 +292,11 @@ def _operation_for_campaign(obj: Campaign, *, changed_attrs: tuple[str, ...]) ->
def _operation_for_object(obj: object, *, changed_attrs: tuple[str, ...]) -> str | None:
state = inspect(obj)
if state.pending:
return "created"
if state.deleted:
return "deleted"
if not _has_attr_changes(state, changed_attrs):
return None
return "updated"
def _has_attr_changes(state: Any, attrs: tuple[str, ...]) -> bool:
return any(name in state.attrs and state.attrs[name].history.has_changes() for name in attrs)
def _previous_value(obj: object, attr_name: str) -> str | None:
state = inspect(obj)
if attr_name not in state.attrs:
return None
history = state.attrs[attr_name].history
if not history.has_changes() or not history.deleted:
return None
value = history.deleted[0]
return str(value) if value is not None else None
return operation_for_object(obj, changed_attrs=changed_attrs)
def _ensure_id(obj: object) -> str:
resource_id = getattr(obj, "id", None)
if resource_id:
return str(resource_id)
resource_id = new_uuid()
setattr(obj, "id", resource_id)
return resource_id
return ensure_object_id(obj, new_uuid)
def _campaign_payload(campaign: Campaign | None) -> dict[str, Any]:
@@ -326,12 +306,12 @@ def _campaign_payload(campaign: Campaign | None) -> dict[str, Any]:
"campaign_id": campaign.id,
"owner_user_id": campaign.owner_user_id,
"owner_group_id": campaign.owner_group_id,
"previous_owner_user_id": _previous_value(campaign, "owner_user_id"),
"previous_owner_group_id": _previous_value(campaign, "owner_group_id"),
"previous_owner_user_id": previous_value(campaign, "owner_user_id"),
"previous_owner_group_id": previous_value(campaign, "owner_group_id"),
"status": campaign.status,
"previous_status": _previous_value(campaign, "status"),
"previous_status": previous_value(campaign, "status"),
"current_version_id": campaign.current_version_id,
"previous_current_version_id": _previous_value(campaign, "current_version_id"),
"previous_current_version_id": previous_value(campaign, "current_version_id"),
}