1363 lines
44 KiB
Python
1363 lines
44 KiB
Python
from __future__ import annotations
|
|
|
|
import dataclasses
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, Query, Response, status
|
|
from sqlalchemy import or_
|
|
from sqlalchemy.orm import Session
|
|
|
|
from govoplan_campaign.backend.schemas import (
|
|
CampaignDeltaResponse,
|
|
CampaignCreateRequest,
|
|
CampaignUpdateRequest,
|
|
CampaignCreateResponse,
|
|
CampaignCreateMinimalRequest,
|
|
CampaignAddressLookupCandidate,
|
|
CampaignAddressLookupResponse,
|
|
CampaignPostboxCatalogResponse,
|
|
CampaignRecipientAddressSource,
|
|
CampaignRecipientAddressSourcesResponse,
|
|
CampaignRecipientAddressSourceSnapshotRequest,
|
|
CampaignRecipientAddressSourceSnapshotResponse,
|
|
CampaignRecipientSnapshotItem,
|
|
RecipientImportMappingProfileListResponse,
|
|
RecipientImportMappingProfilePayload,
|
|
RecipientImportMappingProfileResponse,
|
|
CampaignListResponse,
|
|
CampaignResponse,
|
|
CampaignVersionDetailResponse,
|
|
CampaignVersionResponse,
|
|
CampaignWorkspaceDeltaResponse,
|
|
CampaignWorkspaceResponse,
|
|
)
|
|
from govoplan_core.auth import ApiPrincipal, has_scope, require_scope
|
|
from govoplan_core.audit.logging import audit_from_principal
|
|
from govoplan_core.core.change_sequence import (
|
|
decode_sequence_watermark,
|
|
encode_sequence_watermark,
|
|
max_sequence_id,
|
|
sequence_entries_since,
|
|
sequence_watermark_is_expired,
|
|
)
|
|
from govoplan_core.api.v1.schemas import DeltaDeletedItem
|
|
from govoplan_campaign.backend.change_tracking import (
|
|
CAMPAIGNS_COLLECTION,
|
|
CAMPAIGNS_MODULE_ID,
|
|
CAMPAIGN_ATTEMPTS_COLLECTION,
|
|
CAMPAIGN_ISSUES_COLLECTION,
|
|
CAMPAIGN_JOBS_COLLECTION,
|
|
CAMPAIGN_VERSIONS_COLLECTION,
|
|
)
|
|
from govoplan_campaign.backend.db.models import (
|
|
Campaign,
|
|
CampaignJob,
|
|
CampaignVersion,
|
|
RecipientImportMappingProfile,
|
|
)
|
|
from govoplan_campaign.backend.campaign.postbox_targets import (
|
|
delivery_catalog_payload,
|
|
)
|
|
from govoplan_campaign.backend.integrations import (
|
|
PostboxDeliveryUnavailable,
|
|
postbox_integration,
|
|
)
|
|
from govoplan_core.db.session import get_session
|
|
from govoplan_campaign.backend.reports.campaigns import (
|
|
CampaignReportError,
|
|
generate_campaign_report,
|
|
)
|
|
from govoplan_campaign.backend.report_privacy_policy import (
|
|
CampaignReportPrivacyPolicyError,
|
|
)
|
|
from govoplan_campaign.backend.reports.aggregate import (
|
|
AggregateCampaignReport,
|
|
AggregateCampaignReportError,
|
|
AggregateReportCampaignList,
|
|
aggregate_report_campaign_item,
|
|
generate_aggregate_campaign_report,
|
|
)
|
|
from govoplan_campaign.backend.persistence.campaigns import (
|
|
CampaignPersistenceError,
|
|
create_campaign_version_from_json,
|
|
)
|
|
from govoplan_campaign.backend.persistence.versions import (
|
|
create_minimal_campaign,
|
|
)
|
|
|
|
|
|
from govoplan_campaign.backend.route_support import (
|
|
_apply_recipient_import_profile_payload,
|
|
_campaign_query_for_principal,
|
|
_campaign_response_context,
|
|
_capability_payload,
|
|
_get_campaign_for_principal,
|
|
_get_recipient_import_profile_for_principal,
|
|
_principal_group_ids,
|
|
_registry_capability,
|
|
_require_mail_profile_use_if_needed,
|
|
_require_permission,
|
|
_sync_campaign_metadata_to_current_version,
|
|
_write_current_version_snapshot_if_available,
|
|
bounded_query_rows as _bounded_query_rows,
|
|
)
|
|
|
|
router = APIRouter(prefix="/campaigns", tags=["campaigns"])
|
|
|
|
CAPABILITY_ADDRESSES_LOOKUP = "addresses.lookup"
|
|
CAPABILITY_ADDRESSES_RECIPIENT_SOURCE = "addresses.recipient_source"
|
|
|
|
|
|
@router.post("", response_model=CampaignCreateResponse)
|
|
def create_campaign(
|
|
payload: CampaignCreateRequest,
|
|
session: Session = Depends(get_session),
|
|
principal: ApiPrincipal = Depends(require_scope("campaigns:campaign:create")),
|
|
):
|
|
try:
|
|
if payload.config.get("entries") or payload.config.get("recipients"):
|
|
_require_permission(principal, "campaigns:recipient:write")
|
|
_require_mail_profile_use_if_needed(principal, payload.config)
|
|
campaign, version = create_campaign_version_from_json(
|
|
session,
|
|
tenant_id=principal.tenant_id,
|
|
user_id=principal.user.id,
|
|
raw_json=payload.config,
|
|
source_filename=payload.source_filename,
|
|
source_base_path=payload.source_base_path,
|
|
commit=False,
|
|
)
|
|
audit_from_principal(
|
|
session,
|
|
principal,
|
|
action="campaign.created",
|
|
object_type="campaign",
|
|
object_id=campaign.id,
|
|
details={"version_id": version.id, "external_id": campaign.external_id},
|
|
commit=True,
|
|
)
|
|
_write_current_version_snapshot_if_available(version)
|
|
except HTTPException:
|
|
session.rollback()
|
|
raise
|
|
except Exception as exc:
|
|
session.rollback()
|
|
raise HTTPException(
|
|
status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, detail=str(exc)
|
|
) from exc
|
|
return CampaignCreateResponse(
|
|
campaign=CampaignResponse.model_validate(campaign),
|
|
version=CampaignVersionResponse.model_validate(
|
|
version,
|
|
context=_campaign_response_context(principal),
|
|
),
|
|
)
|
|
|
|
|
|
@router.post("/new", response_model=CampaignCreateResponse)
|
|
def create_minimal_campaign_endpoint(
|
|
payload: CampaignCreateMinimalRequest,
|
|
session: Session = Depends(get_session),
|
|
principal: ApiPrincipal = Depends(require_scope("campaigns:campaign:create")),
|
|
):
|
|
"""Create a minimal editable campaign/version for the WebUI wizard.
|
|
|
|
This is intentionally different from importing a complete campaign JSON. It
|
|
returns a normal Campaign + CampaignVersion whose version is a working copy
|
|
and can be autosaved while incomplete.
|
|
"""
|
|
|
|
try:
|
|
campaign, version = create_minimal_campaign(
|
|
session,
|
|
tenant_id=principal.tenant_id,
|
|
user_id=principal.user.id,
|
|
external_id=payload.external_id,
|
|
name=payload.name,
|
|
description=payload.description,
|
|
current_flow=payload.current_flow,
|
|
current_step=payload.current_step,
|
|
commit=False,
|
|
)
|
|
audit_from_principal(
|
|
session,
|
|
principal,
|
|
action="campaign.created_minimal",
|
|
object_type="campaign",
|
|
object_id=campaign.id,
|
|
details={"version_id": version.id, "external_id": campaign.external_id},
|
|
commit=True,
|
|
)
|
|
_write_current_version_snapshot_if_available(version)
|
|
return CampaignCreateResponse(
|
|
campaign=CampaignResponse.model_validate(campaign),
|
|
version=CampaignVersionResponse.model_validate(
|
|
version,
|
|
context=_campaign_response_context(principal),
|
|
),
|
|
)
|
|
except CampaignPersistenceError as exc:
|
|
session.rollback()
|
|
raise HTTPException(
|
|
status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, detail=str(exc)
|
|
) from exc
|
|
except Exception:
|
|
session.rollback()
|
|
raise
|
|
|
|
|
|
@router.get("", response_model=CampaignListResponse)
|
|
def list_campaigns(
|
|
limit: int = 200,
|
|
session: Session = Depends(get_session),
|
|
principal: ApiPrincipal = Depends(require_scope("campaigns:campaign:read")),
|
|
):
|
|
campaigns = (
|
|
_campaign_query_for_principal(session, principal)
|
|
.order_by(Campaign.updated_at.desc())
|
|
.limit(max(1, min(limit, 500)))
|
|
.all()
|
|
)
|
|
return CampaignListResponse(
|
|
campaigns=[CampaignResponse.model_validate(item) for item in campaigns]
|
|
)
|
|
|
|
|
|
_CAMPAIGN_LIST_DELTA_COLLECTIONS = (CAMPAIGNS_COLLECTION,)
|
|
_CAMPAIGN_FULL_CURSOR_PREFIX = "full:campaigns:"
|
|
_CAMPAIGN_WORKSPACE_DELTA_COLLECTIONS = (
|
|
CAMPAIGNS_COLLECTION,
|
|
CAMPAIGN_VERSIONS_COLLECTION,
|
|
CAMPAIGN_JOBS_COLLECTION,
|
|
CAMPAIGN_ISSUES_COLLECTION,
|
|
CAMPAIGN_ATTEMPTS_COLLECTION,
|
|
)
|
|
|
|
|
|
@dataclasses.dataclass(frozen=True, slots=True)
|
|
class _WorkspaceDeltaState:
|
|
entries: list[object]
|
|
relevant_entries: list[object]
|
|
has_more: bool
|
|
changed_campaign: bool
|
|
changed_version_ids: set[str]
|
|
selected_version_id: str | None
|
|
changed_current_version: bool
|
|
summary_invalidated: bool
|
|
|
|
|
|
def _campaign_delta_watermark(
|
|
session: Session, tenant_id: str, collections: tuple[str, ...]
|
|
) -> str:
|
|
return encode_sequence_watermark(
|
|
max_sequence_id(
|
|
session,
|
|
tenant_id=tenant_id,
|
|
module_id=CAMPAIGNS_MODULE_ID,
|
|
collections=collections,
|
|
)
|
|
)
|
|
|
|
|
|
def _campaign_entry_matches_principal(
|
|
session: Session, principal: ApiPrincipal, payload: dict[str, object]
|
|
) -> bool:
|
|
if has_scope(principal, "tenant:*"):
|
|
return True
|
|
user_ids = {
|
|
value
|
|
for value in (
|
|
payload.get("owner_user_id"),
|
|
payload.get("previous_owner_user_id"),
|
|
)
|
|
if isinstance(value, str)
|
|
}
|
|
if principal.user.id in user_ids:
|
|
return True
|
|
group_ids = _principal_group_ids(session, principal)
|
|
owner_group_ids = {
|
|
value
|
|
for value in (
|
|
payload.get("owner_group_id"),
|
|
payload.get("previous_owner_group_id"),
|
|
)
|
|
if isinstance(value, str)
|
|
}
|
|
if owner_group_ids.intersection(group_ids):
|
|
return True
|
|
share_target_type = payload.get("share_target_type")
|
|
share_target_id = payload.get("share_target_id")
|
|
if share_target_type == "user" and share_target_id == principal.user.id:
|
|
return True
|
|
if (
|
|
share_target_type == "group"
|
|
and isinstance(share_target_id, str)
|
|
and share_target_id in group_ids
|
|
):
|
|
return True
|
|
return False
|
|
|
|
|
|
def _campaign_full_cursor(
|
|
*,
|
|
page: int,
|
|
snapshot_sequence: int,
|
|
) -> str:
|
|
return f"{_CAMPAIGN_FULL_CURSOR_PREFIX}{int(page)}:{int(snapshot_sequence)}"
|
|
|
|
|
|
def _decode_campaign_full_cursor(
|
|
value: str | None,
|
|
) -> tuple[int, int] | None:
|
|
if not value or not value.startswith(_CAMPAIGN_FULL_CURSOR_PREFIX):
|
|
return None
|
|
parts = value[len(_CAMPAIGN_FULL_CURSOR_PREFIX) :].split(":", 1)
|
|
if len(parts) != 2:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail="Invalid campaign full snapshot cursor",
|
|
)
|
|
try:
|
|
page, snapshot_sequence = (int(item) for item in parts)
|
|
except ValueError as exc:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail="Invalid campaign full snapshot cursor",
|
|
) from exc
|
|
if page < 1 or snapshot_sequence < 0:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail="Invalid campaign full snapshot cursor",
|
|
)
|
|
return page, snapshot_sequence
|
|
|
|
|
|
def _campaign_list_full_delta_response(
|
|
session: Session,
|
|
principal: ApiPrincipal,
|
|
*,
|
|
cursor: tuple[int, int] | None = None,
|
|
limit: int = 500,
|
|
) -> CampaignDeltaResponse:
|
|
page = cursor[0] if cursor is not None else 1
|
|
snapshot_sequence = (
|
|
cursor[1]
|
|
if cursor is not None
|
|
else decode_sequence_watermark(
|
|
_campaign_delta_watermark(
|
|
session,
|
|
principal.tenant_id,
|
|
_CAMPAIGN_LIST_DELTA_COLLECTIONS,
|
|
),
|
|
)
|
|
)
|
|
query = _campaign_query_for_principal(session, principal)
|
|
total = query.order_by(None).count()
|
|
pages = max(1, (total + limit - 1) // limit)
|
|
campaigns = (
|
|
query.order_by(Campaign.updated_at.desc(), Campaign.id.asc())
|
|
.offset((page - 1) * limit)
|
|
.limit(limit)
|
|
.all()
|
|
)
|
|
has_more = page < pages
|
|
return CampaignDeltaResponse(
|
|
campaigns=[CampaignResponse.model_validate(item) for item in campaigns],
|
|
deleted=[],
|
|
watermark=(
|
|
_campaign_full_cursor(
|
|
page=page + 1,
|
|
snapshot_sequence=snapshot_sequence,
|
|
)
|
|
if has_more
|
|
else encode_sequence_watermark(snapshot_sequence)
|
|
),
|
|
has_more=has_more,
|
|
full=True,
|
|
total=total,
|
|
page=page,
|
|
page_size=limit,
|
|
pages=pages,
|
|
)
|
|
|
|
|
|
def _campaign_list_delta_response(
|
|
session: Session, principal: ApiPrincipal, *, since: str, limit: int
|
|
) -> CampaignDeltaResponse:
|
|
try:
|
|
since_sequence = decode_sequence_watermark(since)
|
|
except ValueError as exc:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST, detail=str(exc)
|
|
) from exc
|
|
if sequence_watermark_is_expired(
|
|
session,
|
|
since=since_sequence,
|
|
tenant_id=principal.tenant_id,
|
|
module_id=CAMPAIGNS_MODULE_ID,
|
|
collections=_CAMPAIGN_LIST_DELTA_COLLECTIONS,
|
|
):
|
|
return _campaign_list_full_delta_response(
|
|
session,
|
|
principal,
|
|
limit=limit,
|
|
)
|
|
|
|
entries_plus_one = sequence_entries_since(
|
|
session,
|
|
since=since_sequence,
|
|
tenant_id=principal.tenant_id,
|
|
module_id=CAMPAIGNS_MODULE_ID,
|
|
collections=_CAMPAIGN_LIST_DELTA_COLLECTIONS,
|
|
limit=limit + 1,
|
|
)
|
|
has_more = len(entries_plus_one) > limit
|
|
entries = entries_plus_one[:limit]
|
|
changed_ids = list(
|
|
dict.fromkeys(
|
|
entry.resource_id for entry in entries if entry.resource_type == "campaign"
|
|
)
|
|
)
|
|
visible_campaigns = {
|
|
campaign.id: campaign
|
|
for campaign in (
|
|
_campaign_query_for_principal(session, principal)
|
|
.filter(Campaign.id.in_(changed_ids))
|
|
.order_by(Campaign.updated_at.desc())
|
|
.all()
|
|
if changed_ids
|
|
else []
|
|
)
|
|
}
|
|
deleted: dict[tuple[str, str], DeltaDeletedItem] = {}
|
|
for entry in entries:
|
|
if entry.resource_type != "campaign" or entry.resource_id in visible_campaigns:
|
|
continue
|
|
payload = entry.payload or {}
|
|
if not _campaign_entry_matches_principal(session, principal, payload):
|
|
continue
|
|
deleted[(entry.resource_type, entry.resource_id)] = DeltaDeletedItem(
|
|
id=entry.resource_id,
|
|
resource_type=entry.resource_type,
|
|
revision=encode_sequence_watermark(entry.id),
|
|
deleted_at=entry.created_at if entry.operation == "deleted" else None,
|
|
)
|
|
|
|
watermark = (
|
|
encode_sequence_watermark(entries[-1].id)
|
|
if has_more and entries
|
|
else _campaign_delta_watermark(
|
|
session, principal.tenant_id, _CAMPAIGN_LIST_DELTA_COLLECTIONS
|
|
)
|
|
)
|
|
return CampaignDeltaResponse(
|
|
campaigns=[
|
|
CampaignResponse.model_validate(item) for item in visible_campaigns.values()
|
|
],
|
|
deleted=list(deleted.values()),
|
|
watermark=watermark,
|
|
has_more=has_more,
|
|
full=False,
|
|
total=len(visible_campaigns),
|
|
page=1,
|
|
page_size=limit,
|
|
pages=1,
|
|
)
|
|
|
|
|
|
@router.get("/delta", response_model=CampaignDeltaResponse)
|
|
def list_campaigns_delta(
|
|
since: str | None = None,
|
|
limit: int = Query(default=500, ge=1, le=1000),
|
|
session: Session = Depends(get_session),
|
|
principal: ApiPrincipal = Depends(require_scope("campaigns:campaign:read")),
|
|
):
|
|
full_cursor = _decode_campaign_full_cursor(since)
|
|
if since is None or full_cursor is not None:
|
|
return _campaign_list_full_delta_response(
|
|
session,
|
|
principal,
|
|
cursor=full_cursor,
|
|
limit=limit,
|
|
)
|
|
return _campaign_list_delta_response(session, principal, since=since, limit=limit)
|
|
|
|
|
|
@router.get(
|
|
"/recipient-import/mapping-profiles",
|
|
response_model=RecipientImportMappingProfileListResponse,
|
|
)
|
|
def list_recipient_import_mapping_profiles(
|
|
limit: int = Query(default=200, ge=1, le=1000),
|
|
session: Session = Depends(get_session),
|
|
principal: ApiPrincipal = Depends(require_scope("campaigns:recipient:import")),
|
|
):
|
|
profiles = _bounded_query_rows(
|
|
session.query(RecipientImportMappingProfile)
|
|
.filter(
|
|
RecipientImportMappingProfile.tenant_id == principal.tenant_id,
|
|
RecipientImportMappingProfile.owner_user_id == principal.user.id,
|
|
)
|
|
.order_by(
|
|
RecipientImportMappingProfile.updated_at.desc(),
|
|
RecipientImportMappingProfile.name.asc(),
|
|
),
|
|
limit=limit,
|
|
label="Recipient import mapping profile list",
|
|
)
|
|
return RecipientImportMappingProfileListResponse(
|
|
profiles=[
|
|
RecipientImportMappingProfileResponse.model_validate(profile)
|
|
for profile in profiles
|
|
]
|
|
)
|
|
|
|
|
|
@router.post(
|
|
"/recipient-import/mapping-profiles",
|
|
response_model=RecipientImportMappingProfileResponse,
|
|
status_code=status.HTTP_201_CREATED,
|
|
)
|
|
def create_recipient_import_mapping_profile(
|
|
payload: RecipientImportMappingProfilePayload,
|
|
session: Session = Depends(get_session),
|
|
principal: ApiPrincipal = Depends(require_scope("campaigns:recipient:import")),
|
|
):
|
|
profile = RecipientImportMappingProfile(
|
|
tenant_id=principal.tenant_id,
|
|
owner_user_id=principal.user.id,
|
|
name=payload.name.strip(),
|
|
column_count=payload.column_count,
|
|
headers=list(payload.headers),
|
|
normalized_headers=list(payload.normalized_headers),
|
|
ordered_header_fingerprint=payload.ordered_header_fingerprint,
|
|
unordered_header_fingerprint=payload.unordered_header_fingerprint,
|
|
delimiter=payload.delimiter,
|
|
header_rows=payload.header_rows,
|
|
quoted=payload.quoted,
|
|
value_separators=payload.value_separators,
|
|
mappings=[mapping.model_dump(mode="json") for mapping in payload.mappings],
|
|
)
|
|
session.add(profile)
|
|
session.flush()
|
|
audit_from_principal(
|
|
session,
|
|
principal,
|
|
action="campaign.recipient_import_mapping_profile_created",
|
|
object_type="recipient_import_mapping_profile",
|
|
object_id=profile.id,
|
|
details={
|
|
"name": profile.name,
|
|
"ordered_header_fingerprint": profile.ordered_header_fingerprint,
|
|
},
|
|
commit=True,
|
|
)
|
|
session.refresh(profile)
|
|
return RecipientImportMappingProfileResponse.model_validate(profile)
|
|
|
|
|
|
@router.put(
|
|
"/recipient-import/mapping-profiles/{profile_id}",
|
|
response_model=RecipientImportMappingProfileResponse,
|
|
)
|
|
def update_recipient_import_mapping_profile(
|
|
profile_id: str,
|
|
payload: RecipientImportMappingProfilePayload,
|
|
session: Session = Depends(get_session),
|
|
principal: ApiPrincipal = Depends(require_scope("campaigns:recipient:import")),
|
|
):
|
|
profile = _get_recipient_import_profile_for_principal(
|
|
session, profile_id, principal
|
|
)
|
|
_apply_recipient_import_profile_payload(profile, payload)
|
|
session.add(profile)
|
|
audit_from_principal(
|
|
session,
|
|
principal,
|
|
action="campaign.recipient_import_mapping_profile_updated",
|
|
object_type="recipient_import_mapping_profile",
|
|
object_id=profile.id,
|
|
details={
|
|
"name": profile.name,
|
|
"ordered_header_fingerprint": profile.ordered_header_fingerprint,
|
|
},
|
|
commit=True,
|
|
)
|
|
session.refresh(profile)
|
|
return RecipientImportMappingProfileResponse.model_validate(profile)
|
|
|
|
|
|
@router.delete(
|
|
"/recipient-import/mapping-profiles/{profile_id}",
|
|
status_code=status.HTTP_204_NO_CONTENT,
|
|
)
|
|
def delete_recipient_import_mapping_profile(
|
|
profile_id: str,
|
|
session: Session = Depends(get_session),
|
|
principal: ApiPrincipal = Depends(require_scope("campaigns:recipient:import")),
|
|
):
|
|
profile = _get_recipient_import_profile_for_principal(
|
|
session, profile_id, principal
|
|
)
|
|
details = {
|
|
"name": profile.name,
|
|
"ordered_header_fingerprint": profile.ordered_header_fingerprint,
|
|
}
|
|
session.delete(profile)
|
|
audit_from_principal(
|
|
session,
|
|
principal,
|
|
action="campaign.recipient_import_mapping_profile_deleted",
|
|
object_type="recipient_import_mapping_profile",
|
|
object_id=profile_id,
|
|
details=details,
|
|
commit=True,
|
|
)
|
|
return Response(status_code=status.HTTP_204_NO_CONTENT)
|
|
|
|
|
|
@router.get("/aggregate-reports", response_model=AggregateReportCampaignList)
|
|
def list_aggregate_campaign_reports(
|
|
limit: int = Query(default=500, ge=1, le=1000),
|
|
session: Session = Depends(get_session),
|
|
principal: ApiPrincipal = Depends(require_scope("campaigns:report:read")),
|
|
):
|
|
"""List only the business metadata needed to select an aggregate report."""
|
|
|
|
campaigns = _bounded_query_rows(
|
|
_campaign_query_for_principal(session, principal).order_by(
|
|
Campaign.updated_at.desc(), Campaign.id.asc()
|
|
),
|
|
limit=limit,
|
|
label="Aggregate report campaign list",
|
|
)
|
|
return AggregateReportCampaignList(
|
|
campaigns=[aggregate_report_campaign_item(campaign) for campaign in campaigns]
|
|
)
|
|
|
|
|
|
@router.get("/aggregate-reports/{campaign_id}", response_model=AggregateCampaignReport)
|
|
def aggregate_campaign_report(
|
|
campaign_id: str,
|
|
version_id: str | None = None,
|
|
session: Session = Depends(get_session),
|
|
principal: ApiPrincipal = Depends(require_scope("campaigns:report:read")),
|
|
):
|
|
"""Return the privacy-safe aggregate projection without recipient detail."""
|
|
|
|
_get_campaign_for_principal(session, campaign_id, principal)
|
|
try:
|
|
return generate_aggregate_campaign_report(
|
|
session,
|
|
tenant_id=principal.tenant_id,
|
|
campaign_id=campaign_id,
|
|
version_id=version_id,
|
|
)
|
|
except AggregateCampaignReportError as exc:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Campaign report not found",
|
|
) from exc
|
|
except CampaignReportPrivacyPolicyError as exc:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
|
|
detail="Campaign report privacy policy is invalid",
|
|
) from exc
|
|
|
|
|
|
@router.get(
|
|
"/{campaign_id}/address-lookup", response_model=CampaignAddressLookupResponse
|
|
)
|
|
def lookup_campaign_addresses(
|
|
campaign_id: str,
|
|
query: str = Query(default="", min_length=0),
|
|
limit: int = Query(default=25, ge=1, le=100),
|
|
session: Session = Depends(get_session),
|
|
principal: ApiPrincipal = Depends(require_scope("campaigns:recipient:read")),
|
|
):
|
|
_get_campaign_for_principal(session, campaign_id, principal)
|
|
capability = _registry_capability(CAPABILITY_ADDRESSES_LOOKUP)
|
|
if capability is None or not hasattr(capability, "lookup"):
|
|
return CampaignAddressLookupResponse(available=False, candidates=[])
|
|
candidates = getattr(capability, "lookup")(
|
|
session, principal, query=query, limit=limit
|
|
)
|
|
return CampaignAddressLookupResponse(
|
|
available=True,
|
|
candidates=[
|
|
CampaignAddressLookupCandidate.model_validate(
|
|
_capability_payload(candidate)
|
|
)
|
|
for candidate in candidates
|
|
],
|
|
)
|
|
|
|
|
|
@router.get(
|
|
"/{campaign_id}/recipient-address-sources",
|
|
response_model=CampaignRecipientAddressSourcesResponse,
|
|
)
|
|
def list_campaign_recipient_address_sources(
|
|
campaign_id: str,
|
|
session: Session = Depends(get_session),
|
|
principal: ApiPrincipal = Depends(require_scope("campaigns:recipient:read")),
|
|
):
|
|
_get_campaign_for_principal(session, campaign_id, principal)
|
|
capability = _registry_capability(CAPABILITY_ADDRESSES_RECIPIENT_SOURCE)
|
|
if capability is None or not hasattr(capability, "list_sources"):
|
|
return CampaignRecipientAddressSourcesResponse(available=False, sources=[])
|
|
sources = getattr(capability, "list_sources")(session, principal)
|
|
return CampaignRecipientAddressSourcesResponse(
|
|
available=True,
|
|
sources=[
|
|
CampaignRecipientAddressSource.model_validate(_capability_payload(source))
|
|
for source in sources
|
|
],
|
|
)
|
|
|
|
|
|
@router.get(
|
|
"/{campaign_id}/postbox-catalog",
|
|
response_model=CampaignPostboxCatalogResponse,
|
|
)
|
|
def campaign_postbox_catalog(
|
|
campaign_id: str,
|
|
session: Session = Depends(get_session),
|
|
principal: ApiPrincipal = Depends(require_scope("campaigns:recipient:write")),
|
|
):
|
|
_get_campaign_for_principal(session, campaign_id, principal, write=True)
|
|
integration = postbox_integration()
|
|
if not integration.available:
|
|
return CampaignPostboxCatalogResponse(available=False)
|
|
try:
|
|
payload = delivery_catalog_payload(
|
|
integration.delivery_catalog(
|
|
session,
|
|
tenant_id=principal.tenant_id,
|
|
)
|
|
)
|
|
except PostboxDeliveryUnavailable:
|
|
return CampaignPostboxCatalogResponse(available=False)
|
|
return CampaignPostboxCatalogResponse(
|
|
available=True,
|
|
postboxes=payload.get("postboxes", []),
|
|
templates=payload.get("templates", []),
|
|
organization_units=payload.get("organization_units", []),
|
|
)
|
|
|
|
|
|
@router.post(
|
|
"/{campaign_id}/recipient-address-sources/snapshot",
|
|
response_model=CampaignRecipientAddressSourceSnapshotResponse,
|
|
)
|
|
def snapshot_campaign_recipient_address_source(
|
|
campaign_id: str,
|
|
payload: CampaignRecipientAddressSourceSnapshotRequest,
|
|
session: Session = Depends(get_session),
|
|
principal: ApiPrincipal = Depends(require_scope("campaigns:recipient:import")),
|
|
):
|
|
_get_campaign_for_principal(session, campaign_id, principal, write=True)
|
|
capability = _registry_capability(CAPABILITY_ADDRESSES_RECIPIENT_SOURCE)
|
|
if capability is None or not hasattr(capability, "snapshot"):
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Address recipient-source capability is not available",
|
|
)
|
|
try:
|
|
snapshot = getattr(capability, "snapshot")(
|
|
session, principal, source_id=payload.source_id
|
|
)
|
|
except ValueError as exc:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, detail=str(exc)
|
|
) from exc
|
|
snapshot_payload = _capability_payload(snapshot)
|
|
recipients = [
|
|
CampaignRecipientSnapshotItem.model_validate(_capability_payload(item))
|
|
for item in snapshot_payload.get("recipients", [])
|
|
]
|
|
return CampaignRecipientAddressSourceSnapshotResponse(
|
|
source_id=str(snapshot_payload.get("source_id") or ""),
|
|
source_label=str(snapshot_payload.get("source_label") or ""),
|
|
source_kind=str(snapshot_payload.get("source_kind") or ""),
|
|
source_revision=str(snapshot_payload.get("source_revision") or ""),
|
|
generated_at=str(snapshot_payload.get("generated_at") or ""),
|
|
recipients=recipients,
|
|
provenance=snapshot_payload.get("provenance")
|
|
if isinstance(snapshot_payload.get("provenance"), dict)
|
|
else {},
|
|
)
|
|
|
|
|
|
@router.get("/{campaign_id}", response_model=CampaignResponse)
|
|
def get_campaign(
|
|
campaign_id: str,
|
|
session: Session = Depends(get_session),
|
|
principal: ApiPrincipal = Depends(require_scope("campaigns:campaign:read")),
|
|
):
|
|
return CampaignResponse.model_validate(
|
|
_get_campaign_for_principal(session, campaign_id, principal)
|
|
)
|
|
|
|
|
|
def _campaign_workspace_response(
|
|
session: Session,
|
|
principal: ApiPrincipal,
|
|
*,
|
|
campaign_id: str,
|
|
version_id: str | None,
|
|
include_current_version: bool,
|
|
include_summary: bool,
|
|
include_versions: bool,
|
|
) -> CampaignWorkspaceResponse:
|
|
campaign = _get_campaign_for_principal(session, campaign_id, principal)
|
|
|
|
versions: list[CampaignVersion] = []
|
|
if include_versions or include_current_version:
|
|
versions = _bounded_query_rows(
|
|
session.query(CampaignVersion)
|
|
.filter(CampaignVersion.campaign_id == campaign.id)
|
|
.order_by(CampaignVersion.version_number.desc()),
|
|
limit=1000,
|
|
label="Campaign version history",
|
|
)
|
|
|
|
selected_version_id = (
|
|
version_id
|
|
or campaign.current_version_id
|
|
or (versions[0].id if versions else None)
|
|
)
|
|
current_version: CampaignVersion | None = None
|
|
if include_current_version and selected_version_id:
|
|
_require_permission(principal, "campaigns:recipient:read")
|
|
current_version = (
|
|
session.query(CampaignVersion)
|
|
.filter(
|
|
CampaignVersion.id == selected_version_id,
|
|
CampaignVersion.campaign_id == campaign.id,
|
|
)
|
|
.one_or_none()
|
|
)
|
|
|
|
summary_payload: dict[str, object] | None = None
|
|
if include_summary:
|
|
try:
|
|
summary_payload = generate_campaign_report(
|
|
session,
|
|
tenant_id=principal.tenant_id,
|
|
campaign_id=campaign_id,
|
|
version_id=selected_version_id,
|
|
include_jobs=False,
|
|
include_diagnostics=has_scope(principal, "campaigns:diagnostic:read"),
|
|
)
|
|
except CampaignReportError:
|
|
summary_payload = None
|
|
|
|
return CampaignWorkspaceResponse(
|
|
campaign=CampaignResponse.model_validate(campaign),
|
|
versions=[
|
|
CampaignVersionResponse.model_validate(
|
|
item,
|
|
context=_campaign_response_context(principal),
|
|
)
|
|
for item in versions
|
|
]
|
|
if include_versions
|
|
else [],
|
|
current_version=(
|
|
CampaignVersionDetailResponse.model_validate(
|
|
current_version,
|
|
context=_campaign_response_context(principal),
|
|
)
|
|
if current_version is not None
|
|
else None
|
|
),
|
|
summary=summary_payload,
|
|
selected_version_id=selected_version_id,
|
|
)
|
|
|
|
|
|
@router.get("/{campaign_id}/workspace", response_model=CampaignWorkspaceResponse)
|
|
def get_campaign_workspace(
|
|
campaign_id: str,
|
|
version_id: str | None = None,
|
|
include_current_version: bool = True,
|
|
include_summary: bool = False,
|
|
include_versions: bool = True,
|
|
session: Session = Depends(get_session),
|
|
principal: ApiPrincipal = Depends(require_scope("campaigns:campaign:read")),
|
|
):
|
|
return _campaign_workspace_response(
|
|
session,
|
|
principal,
|
|
campaign_id=campaign_id,
|
|
version_id=version_id,
|
|
include_current_version=include_current_version,
|
|
include_summary=include_summary,
|
|
include_versions=include_versions,
|
|
)
|
|
|
|
|
|
def _workspace_delta_deleted_item(entry) -> DeltaDeletedItem:
|
|
return DeltaDeletedItem(
|
|
id=entry.resource_id,
|
|
resource_type=entry.resource_type,
|
|
revision=encode_sequence_watermark(entry.id),
|
|
deleted_at=entry.created_at if entry.operation == "deleted" else None,
|
|
)
|
|
|
|
|
|
def _entry_belongs_to_campaign(entry, campaign_id: str) -> bool:
|
|
payload = entry.payload or {}
|
|
if payload.get("campaign_id") == campaign_id:
|
|
return True
|
|
return entry.resource_type == "campaign" and entry.resource_id == campaign_id
|
|
|
|
|
|
def _campaign_workspace_full_delta_response(
|
|
session: Session,
|
|
principal: ApiPrincipal,
|
|
*,
|
|
campaign_id: str,
|
|
version_id: str | None,
|
|
include_current_version: bool,
|
|
include_summary: bool,
|
|
include_versions: bool,
|
|
) -> CampaignWorkspaceDeltaResponse:
|
|
payload = _campaign_workspace_response(
|
|
session,
|
|
principal,
|
|
campaign_id=campaign_id,
|
|
version_id=version_id,
|
|
include_current_version=include_current_version,
|
|
include_summary=include_summary,
|
|
include_versions=include_versions,
|
|
)
|
|
return CampaignWorkspaceDeltaResponse(
|
|
**payload.model_dump(),
|
|
deleted=[],
|
|
watermark=_campaign_delta_watermark(
|
|
session, principal.tenant_id, _CAMPAIGN_WORKSPACE_DELTA_COLLECTIONS
|
|
),
|
|
has_more=False,
|
|
full=True,
|
|
)
|
|
|
|
|
|
def _campaign_workspace_delta_response(
|
|
session: Session,
|
|
principal: ApiPrincipal,
|
|
*,
|
|
campaign_id: str,
|
|
version_id: str | None,
|
|
include_current_version: bool,
|
|
include_summary: bool,
|
|
include_versions: bool,
|
|
since: str,
|
|
limit: int,
|
|
) -> CampaignWorkspaceDeltaResponse:
|
|
campaign = _get_campaign_for_principal(session, campaign_id, principal)
|
|
try:
|
|
since_sequence = decode_sequence_watermark(since)
|
|
except ValueError as exc:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST, detail=str(exc)
|
|
) from exc
|
|
if sequence_watermark_is_expired(
|
|
session,
|
|
since=since_sequence,
|
|
tenant_id=principal.tenant_id,
|
|
module_id=CAMPAIGNS_MODULE_ID,
|
|
collections=_CAMPAIGN_WORKSPACE_DELTA_COLLECTIONS,
|
|
):
|
|
return _campaign_workspace_full_delta_response(
|
|
session,
|
|
principal,
|
|
campaign_id=campaign_id,
|
|
version_id=version_id,
|
|
include_current_version=include_current_version,
|
|
include_summary=include_summary,
|
|
include_versions=include_versions,
|
|
)
|
|
|
|
delta = _campaign_workspace_delta_state(
|
|
session,
|
|
principal,
|
|
campaign=campaign,
|
|
version_id=version_id,
|
|
since_sequence=since_sequence,
|
|
limit=limit,
|
|
)
|
|
return CampaignWorkspaceDeltaResponse(
|
|
campaign=CampaignResponse.model_validate(campaign)
|
|
if delta.changed_campaign
|
|
else None,
|
|
versions=_campaign_workspace_delta_versions(
|
|
session,
|
|
principal,
|
|
campaign,
|
|
delta,
|
|
include_versions=include_versions,
|
|
),
|
|
current_version=_campaign_workspace_delta_current_version(
|
|
session,
|
|
principal,
|
|
campaign=campaign,
|
|
delta=delta,
|
|
include_current_version=include_current_version,
|
|
),
|
|
summary=_campaign_workspace_delta_summary(
|
|
session,
|
|
principal,
|
|
campaign_id=campaign_id,
|
|
delta=delta,
|
|
include_summary=include_summary,
|
|
),
|
|
selected_version_id=delta.selected_version_id,
|
|
deleted=_campaign_workspace_delta_deleted(delta.relevant_entries),
|
|
watermark=_campaign_workspace_delta_watermark(session, principal, delta),
|
|
has_more=delta.has_more,
|
|
full=False,
|
|
)
|
|
|
|
|
|
def _campaign_workspace_delta_state(
|
|
session: Session,
|
|
principal: ApiPrincipal,
|
|
*,
|
|
campaign: Campaign,
|
|
version_id: str | None,
|
|
since_sequence: int,
|
|
limit: int,
|
|
) -> _WorkspaceDeltaState:
|
|
entries_plus_one = sequence_entries_since(
|
|
session,
|
|
since=since_sequence,
|
|
tenant_id=principal.tenant_id,
|
|
module_id=CAMPAIGNS_MODULE_ID,
|
|
collections=_CAMPAIGN_WORKSPACE_DELTA_COLLECTIONS,
|
|
limit=limit + 1,
|
|
)
|
|
entries = entries_plus_one[:limit]
|
|
relevant_entries = [
|
|
entry for entry in entries if _entry_belongs_to_campaign(entry, campaign.id)
|
|
]
|
|
changed_version_ids = {
|
|
entry.resource_id
|
|
for entry in relevant_entries
|
|
if entry.resource_type == "campaign_version" and entry.operation != "deleted"
|
|
}
|
|
selected_version_id = version_id or campaign.current_version_id
|
|
return _WorkspaceDeltaState(
|
|
entries=entries,
|
|
relevant_entries=relevant_entries,
|
|
has_more=len(entries_plus_one) > limit,
|
|
changed_campaign=any(
|
|
entry.resource_type == "campaign" for entry in relevant_entries
|
|
),
|
|
changed_version_ids=changed_version_ids,
|
|
selected_version_id=selected_version_id,
|
|
changed_current_version=bool(
|
|
selected_version_id and selected_version_id in changed_version_ids
|
|
),
|
|
summary_invalidated=any(
|
|
entry.collection
|
|
in {
|
|
CAMPAIGN_JOBS_COLLECTION,
|
|
CAMPAIGN_ISSUES_COLLECTION,
|
|
CAMPAIGN_ATTEMPTS_COLLECTION,
|
|
}
|
|
for entry in relevant_entries
|
|
),
|
|
)
|
|
|
|
|
|
def _campaign_workspace_delta_deleted(entries: list[object]) -> list[dict[str, object]]:
|
|
return [
|
|
_workspace_delta_deleted_item(entry)
|
|
for entry in entries
|
|
if entry.operation == "deleted" and entry.resource_type != "campaign"
|
|
]
|
|
|
|
|
|
def _campaign_workspace_delta_versions(
|
|
session: Session,
|
|
principal: ApiPrincipal,
|
|
campaign: Campaign,
|
|
delta: _WorkspaceDeltaState,
|
|
*,
|
|
include_versions: bool,
|
|
) -> list[CampaignVersionResponse]:
|
|
if not include_versions or not delta.changed_version_ids:
|
|
return []
|
|
versions = (
|
|
session.query(CampaignVersion)
|
|
.filter(
|
|
CampaignVersion.campaign_id == campaign.id,
|
|
CampaignVersion.id.in_(delta.changed_version_ids),
|
|
)
|
|
.order_by(CampaignVersion.version_number.desc())
|
|
.all()
|
|
)
|
|
return [
|
|
CampaignVersionResponse.model_validate(
|
|
item,
|
|
context=_campaign_response_context(principal),
|
|
)
|
|
for item in versions
|
|
]
|
|
|
|
|
|
def _campaign_workspace_delta_current_version(
|
|
session: Session,
|
|
principal: ApiPrincipal,
|
|
*,
|
|
campaign: Campaign,
|
|
delta: _WorkspaceDeltaState,
|
|
include_current_version: bool,
|
|
) -> CampaignVersionDetailResponse | None:
|
|
if not include_current_version or not delta.selected_version_id:
|
|
return None
|
|
if not (delta.changed_campaign or delta.changed_current_version):
|
|
return None
|
|
_require_permission(principal, "campaigns:recipient:read")
|
|
current_version = (
|
|
session.query(CampaignVersion)
|
|
.filter(
|
|
CampaignVersion.id == delta.selected_version_id,
|
|
CampaignVersion.campaign_id == campaign.id,
|
|
)
|
|
.one_or_none()
|
|
)
|
|
return (
|
|
CampaignVersionDetailResponse.model_validate(
|
|
current_version,
|
|
context=_campaign_response_context(principal),
|
|
)
|
|
if current_version is not None
|
|
else None
|
|
)
|
|
|
|
|
|
def _campaign_workspace_delta_summary(
|
|
session: Session,
|
|
principal: ApiPrincipal,
|
|
*,
|
|
campaign_id: str,
|
|
delta: _WorkspaceDeltaState,
|
|
include_summary: bool,
|
|
) -> dict[str, object] | None:
|
|
if not include_summary:
|
|
return None
|
|
if not (
|
|
delta.changed_campaign or delta.changed_version_ids or delta.summary_invalidated
|
|
):
|
|
return None
|
|
try:
|
|
return generate_campaign_report(
|
|
session,
|
|
tenant_id=principal.tenant_id,
|
|
campaign_id=campaign_id,
|
|
version_id=delta.selected_version_id,
|
|
include_jobs=False,
|
|
include_diagnostics=has_scope(principal, "campaigns:diagnostic:read"),
|
|
)
|
|
except CampaignReportError:
|
|
return None
|
|
|
|
|
|
def _campaign_workspace_delta_watermark(
|
|
session: Session,
|
|
principal: ApiPrincipal,
|
|
delta: _WorkspaceDeltaState,
|
|
) -> str:
|
|
if delta.has_more and delta.entries:
|
|
return encode_sequence_watermark(delta.entries[-1].id)
|
|
return _campaign_delta_watermark(
|
|
session, principal.tenant_id, _CAMPAIGN_WORKSPACE_DELTA_COLLECTIONS
|
|
)
|
|
|
|
|
|
@router.get(
|
|
"/{campaign_id}/workspace/delta", response_model=CampaignWorkspaceDeltaResponse
|
|
)
|
|
def get_campaign_workspace_delta(
|
|
campaign_id: str,
|
|
version_id: str | None = None,
|
|
include_current_version: bool = True,
|
|
include_summary: bool = False,
|
|
include_versions: bool = True,
|
|
since: str | None = None,
|
|
limit: int = Query(default=500, ge=1, le=1000),
|
|
session: Session = Depends(get_session),
|
|
principal: ApiPrincipal = Depends(require_scope("campaigns:campaign:read")),
|
|
):
|
|
if since is None:
|
|
return _campaign_workspace_full_delta_response(
|
|
session,
|
|
principal,
|
|
campaign_id=campaign_id,
|
|
version_id=version_id,
|
|
include_current_version=include_current_version,
|
|
include_summary=include_summary,
|
|
include_versions=include_versions,
|
|
)
|
|
return _campaign_workspace_delta_response(
|
|
session,
|
|
principal,
|
|
campaign_id=campaign_id,
|
|
version_id=version_id,
|
|
include_current_version=include_current_version,
|
|
include_summary=include_summary,
|
|
include_versions=include_versions,
|
|
since=since,
|
|
limit=limit,
|
|
)
|
|
|
|
|
|
@router.put("/{campaign_id}", response_model=CampaignResponse)
|
|
def update_campaign_metadata_endpoint(
|
|
campaign_id: str,
|
|
payload: CampaignUpdateRequest,
|
|
session: Session = Depends(get_session),
|
|
principal: ApiPrincipal = Depends(require_scope("campaigns:campaign:update")),
|
|
):
|
|
campaign = _get_campaign_for_principal(session, campaign_id, principal, write=True)
|
|
if payload.external_id is not None:
|
|
value = payload.external_id.strip()
|
|
if not value:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
|
|
detail="Campaign ID cannot be empty",
|
|
)
|
|
duplicate = (
|
|
session.query(Campaign)
|
|
.filter(
|
|
Campaign.tenant_id == principal.tenant_id,
|
|
Campaign.external_id == value,
|
|
Campaign.id != campaign.id,
|
|
)
|
|
.one_or_none()
|
|
)
|
|
if duplicate:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_409_CONFLICT,
|
|
detail="Campaign ID already exists for this tenant",
|
|
)
|
|
campaign.external_id = value
|
|
if payload.name is not None:
|
|
value = payload.name.strip()
|
|
if not value:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
|
|
detail="Campaign name cannot be empty",
|
|
)
|
|
campaign.name = value
|
|
if payload.status is not None:
|
|
campaign.status = payload.status
|
|
if payload.description is not None:
|
|
campaign.description = payload.description
|
|
|
|
_sync_campaign_metadata_to_current_version(session, campaign)
|
|
session.add(campaign)
|
|
session.flush()
|
|
try:
|
|
audit_from_principal(
|
|
session,
|
|
principal,
|
|
action="campaign.metadata_updated",
|
|
object_type="campaign",
|
|
object_id=campaign.id,
|
|
details={"external_id": campaign.external_id, "name": campaign.name},
|
|
commit=True,
|
|
)
|
|
except Exception:
|
|
session.rollback()
|
|
raise
|
|
session.refresh(campaign)
|
|
if campaign.current_version_id:
|
|
current_version = session.get(CampaignVersion, campaign.current_version_id)
|
|
if current_version is not None:
|
|
_write_current_version_snapshot_if_available(current_version)
|
|
return CampaignResponse.model_validate(campaign)
|
|
|
|
|
|
@router.post("/{campaign_id}/archive", response_model=CampaignResponse)
|
|
def archive_campaign(
|
|
campaign_id: str,
|
|
session: Session = Depends(get_session),
|
|
principal: ApiPrincipal = Depends(require_scope("campaigns:campaign:archive")),
|
|
):
|
|
campaign = _get_campaign_for_principal(session, campaign_id, principal, write=True)
|
|
if campaign.status in {"queued", "sending", "outcome_unknown"}:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_409_CONFLICT,
|
|
detail="Active or uncertain delivery must be resolved before archiving.",
|
|
)
|
|
campaign.status = "archived"
|
|
session.add(campaign)
|
|
audit_from_principal(
|
|
session,
|
|
principal,
|
|
action="campaign.archived",
|
|
object_type="campaign",
|
|
object_id=campaign.id,
|
|
details={},
|
|
commit=True,
|
|
)
|
|
session.refresh(campaign)
|
|
return CampaignResponse.model_validate(campaign)
|
|
|
|
|
|
@router.delete("/{campaign_id}", status_code=status.HTTP_204_NO_CONTENT)
|
|
def delete_draft_campaign(
|
|
campaign_id: str,
|
|
session: Session = Depends(get_session),
|
|
principal: ApiPrincipal = Depends(require_scope("campaigns:campaign:delete")),
|
|
):
|
|
campaign = _get_campaign_for_principal(session, campaign_id, principal, write=True)
|
|
if campaign.status != "draft":
|
|
raise HTTPException(
|
|
status_code=status.HTTP_409_CONFLICT,
|
|
detail="Only untouched draft campaigns can be deleted.",
|
|
)
|
|
if (
|
|
session.query(CampaignJob.id)
|
|
.filter(CampaignJob.campaign_id == campaign.id)
|
|
.first()
|
|
is not None
|
|
):
|
|
raise HTTPException(
|
|
status_code=status.HTTP_409_CONFLICT,
|
|
detail="Campaigns with built or delivery jobs must be archived instead of deleted.",
|
|
)
|
|
protected_version = (
|
|
session.query(CampaignVersion.id)
|
|
.filter(
|
|
CampaignVersion.campaign_id == campaign.id,
|
|
or_(
|
|
CampaignVersion.locked_at.is_not(None),
|
|
CampaignVersion.user_lock_state.is_not(None),
|
|
CampaignVersion.published_at.is_not(None),
|
|
CampaignVersion.execution_snapshot_at.is_not(None),
|
|
),
|
|
)
|
|
.first()
|
|
)
|
|
if protected_version is not None:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_409_CONFLICT,
|
|
detail="Audit-relevant campaign versions must be archived instead of deleted.",
|
|
)
|
|
campaign.status = "deleted"
|
|
session.add(campaign)
|
|
audit_from_principal(
|
|
session,
|
|
principal,
|
|
action="campaign.deleted",
|
|
object_type="campaign",
|
|
object_id=campaign.id,
|
|
details={"mode": "soft"},
|
|
commit=True,
|
|
)
|
|
return Response(status_code=status.HTTP_204_NO_CONTENT)
|