723 lines
25 KiB
Python
723 lines
25 KiB
Python
from __future__ import annotations
|
|
|
|
from collections.abc import Iterable, Mapping
|
|
|
|
from fastapi import HTTPException
|
|
from sqlalchemy import and_, or_
|
|
|
|
from govoplan_core.core.access import AccessDecisionProvenance, PrincipalRef
|
|
from govoplan_core.core.campaigns import (
|
|
CampaignAccessProvider,
|
|
CampaignDeliveryTaskProvider,
|
|
CampaignMailPolicyContext,
|
|
CampaignMailPolicyContextProvider,
|
|
CampaignPolicyContext,
|
|
CampaignPolicyContextProvider,
|
|
CampaignRetentionProvider,
|
|
)
|
|
from govoplan_core.core.ownership import (
|
|
OwnershipActionDecision,
|
|
OwnershipSubjectRef,
|
|
OwnershipTransferError,
|
|
)
|
|
from govoplan_core.security.module_permissions import scopes_grant_compatible
|
|
|
|
from govoplan_campaign.backend.db.models import (
|
|
Campaign,
|
|
CampaignJob,
|
|
CampaignShare,
|
|
CampaignVersion,
|
|
)
|
|
|
|
|
|
READ_ACTIONS = {"campaigns:campaign:read", "campaign:read"}
|
|
CAMPAIGN_RESOURCE_TYPES = {
|
|
"campaign",
|
|
"campaign_object",
|
|
"campaigns:campaign",
|
|
}
|
|
CAMPAIGN_VERSION_RESOURCE_TYPES = {
|
|
"campaign_version",
|
|
"campaigns:version",
|
|
}
|
|
CAMPAIGN_DELIVERY_JOB_RESOURCE_TYPES = {
|
|
"campaign_delivery_job",
|
|
"campaign_job",
|
|
"campaigns:delivery_job",
|
|
}
|
|
CAMPAIGN_REPORT_RESOURCE_TYPES = {
|
|
"campaign_report",
|
|
"campaigns:report",
|
|
}
|
|
|
|
|
|
def campaign_report_resource_id(
|
|
*,
|
|
campaign_id: str,
|
|
version_id: str,
|
|
report_kind: str,
|
|
) -> str:
|
|
kind = report_kind.strip().lower()
|
|
if not campaign_id or not version_id or not kind or ":" in kind:
|
|
raise ValueError("Campaign report references require campaign, version, and a bounded kind")
|
|
return f"{campaign_id}:{version_id}:{kind}"
|
|
|
|
|
|
def _campaign_report_reference(
|
|
resource_id: str,
|
|
) -> tuple[str, str, str] | None:
|
|
parts = resource_id.split(":", 2)
|
|
if len(parts) != 3 or not all(part.strip() for part in parts):
|
|
return None
|
|
return parts[0], parts[1], parts[2].strip().lower()
|
|
|
|
|
|
class CampaignMailPolicyContextService(CampaignMailPolicyContextProvider):
|
|
def get_campaign_mail_policy_context(
|
|
self,
|
|
session: object,
|
|
*,
|
|
tenant_id: str,
|
|
campaign_id: str,
|
|
) -> CampaignMailPolicyContext | None:
|
|
campaign = session.get(Campaign, campaign_id) # type: ignore[attr-defined]
|
|
if campaign is None or campaign.tenant_id != tenant_id:
|
|
return None
|
|
return CampaignMailPolicyContext(
|
|
id=campaign.id,
|
|
tenant_id=campaign.tenant_id,
|
|
owner_user_id=campaign.owner_user_id,
|
|
owner_group_id=campaign.owner_group_id,
|
|
mail_profile_policy=dict(campaign.mail_profile_policy or {}),
|
|
)
|
|
|
|
def set_campaign_mail_profile_policy(
|
|
self,
|
|
session: object,
|
|
*,
|
|
tenant_id: str,
|
|
campaign_id: str,
|
|
policy: Mapping[str, object],
|
|
) -> Mapping[str, object]:
|
|
campaign = session.get(Campaign, campaign_id) # type: ignore[attr-defined]
|
|
if campaign is None or campaign.tenant_id != tenant_id:
|
|
raise ValueError("Campaign policy not found")
|
|
payload = dict(policy)
|
|
campaign.mail_profile_policy = payload
|
|
session.add(campaign) # type: ignore[attr-defined]
|
|
return payload
|
|
|
|
|
|
def mail_policy_context_capability(context: object) -> CampaignMailPolicyContextService:
|
|
return CampaignMailPolicyContextService()
|
|
|
|
|
|
class CampaignAccessService(CampaignAccessProvider):
|
|
def campaign_exists(self, session: object, *, tenant_id: str, campaign_id: str) -> bool:
|
|
campaign = session.get(Campaign, campaign_id) # type: ignore[attr-defined]
|
|
return bool(campaign and campaign.tenant_id == tenant_id)
|
|
|
|
def can_read_campaign(
|
|
self,
|
|
session: object,
|
|
*,
|
|
tenant_id: str,
|
|
campaign_id: str,
|
|
user_id: str,
|
|
group_ids: Iterable[str] = (),
|
|
tenant_admin: bool = False,
|
|
) -> bool:
|
|
campaign = session.get(Campaign, campaign_id) # type: ignore[attr-defined]
|
|
if campaign is None or campaign.tenant_id != tenant_id:
|
|
return False
|
|
if tenant_admin or campaign.owner_user_id == user_id:
|
|
return True
|
|
group_id_set = {str(group_id) for group_id in group_ids}
|
|
if campaign.owner_group_id and campaign.owner_group_id in group_id_set:
|
|
return True
|
|
share_clauses = [
|
|
and_(
|
|
CampaignShare.target_type == "user",
|
|
CampaignShare.target_id == user_id,
|
|
)
|
|
]
|
|
if group_id_set:
|
|
share_clauses.append(
|
|
and_(
|
|
CampaignShare.target_type == "group",
|
|
CampaignShare.target_id.in_(group_id_set),
|
|
)
|
|
)
|
|
share = (
|
|
session.query(CampaignShare) # type: ignore[attr-defined]
|
|
.filter(
|
|
CampaignShare.tenant_id == tenant_id,
|
|
CampaignShare.campaign_id == campaign.id,
|
|
CampaignShare.revoked_at.is_(None),
|
|
or_(*share_clauses),
|
|
)
|
|
.first()
|
|
)
|
|
return share is not None
|
|
|
|
def explain_resource_provenance(
|
|
self,
|
|
session: object,
|
|
principal: PrincipalRef,
|
|
*,
|
|
resource_type: str,
|
|
resource_id: str,
|
|
action: str,
|
|
) -> tuple[AccessDecisionProvenance, ...]:
|
|
normalized_type = resource_type.lower().strip()
|
|
child_item: AccessDecisionProvenance | None = None
|
|
campaign: Campaign | None
|
|
if normalized_type in CAMPAIGN_RESOURCE_TYPES:
|
|
campaign = session.get(Campaign, resource_id) # type: ignore[attr-defined]
|
|
elif normalized_type in CAMPAIGN_VERSION_RESOURCE_TYPES:
|
|
version = session.get(CampaignVersion, resource_id) # type: ignore[attr-defined]
|
|
if version is None:
|
|
return _missing_resource_provenance(
|
|
principal,
|
|
resource_type="campaign_version",
|
|
resource_id=resource_id,
|
|
)
|
|
campaign = session.get(Campaign, version.campaign_id) # type: ignore[attr-defined]
|
|
child_item = AccessDecisionProvenance(
|
|
kind="resource",
|
|
id=version.id,
|
|
label=f"Version {version.version_number}",
|
|
tenant_id=campaign.tenant_id if campaign else principal.tenant_id,
|
|
source="campaigns.version",
|
|
details={
|
|
"resource_type": "campaign_version",
|
|
"campaign_id": version.campaign_id,
|
|
"version_number": version.version_number,
|
|
"workflow_state": version.workflow_state,
|
|
"authorization_inherited_from": {
|
|
"resource_type": "campaign",
|
|
"resource_id": version.campaign_id,
|
|
},
|
|
},
|
|
)
|
|
elif normalized_type in CAMPAIGN_DELIVERY_JOB_RESOURCE_TYPES:
|
|
job = session.get(CampaignJob, resource_id) # type: ignore[attr-defined]
|
|
if job is None:
|
|
return _missing_resource_provenance(
|
|
principal,
|
|
resource_type="campaign_delivery_job",
|
|
resource_id=resource_id,
|
|
)
|
|
campaign = session.get(Campaign, job.campaign_id) # type: ignore[attr-defined]
|
|
child_item = AccessDecisionProvenance(
|
|
kind="resource",
|
|
id=job.id,
|
|
label=job.recipient_email or f"Recipient {job.entry_index}",
|
|
tenant_id=job.tenant_id,
|
|
source="campaigns.delivery_job",
|
|
details={
|
|
"resource_type": "campaign_delivery_job",
|
|
"campaign_id": job.campaign_id,
|
|
"campaign_version_id": job.campaign_version_id,
|
|
"queue_status": job.queue_status,
|
|
"send_status": job.send_status,
|
|
"authorization_inherited_from": {
|
|
"resource_type": "campaign",
|
|
"resource_id": job.campaign_id,
|
|
},
|
|
},
|
|
)
|
|
elif normalized_type in CAMPAIGN_REPORT_RESOURCE_TYPES:
|
|
reference = _campaign_report_reference(resource_id)
|
|
if reference is None:
|
|
return _missing_resource_provenance(
|
|
principal,
|
|
resource_type="campaign_report",
|
|
resource_id=resource_id,
|
|
reason="invalid_reference",
|
|
)
|
|
campaign_id, version_id, report_kind = reference
|
|
campaign = session.get(Campaign, campaign_id) # type: ignore[attr-defined]
|
|
version = session.get(CampaignVersion, version_id) # type: ignore[attr-defined]
|
|
if (
|
|
campaign is None
|
|
or version is None
|
|
or version.campaign_id != campaign.id
|
|
):
|
|
return _missing_resource_provenance(
|
|
principal,
|
|
resource_type="campaign_report",
|
|
resource_id=resource_id,
|
|
)
|
|
child_item = AccessDecisionProvenance(
|
|
kind="resource",
|
|
id=resource_id,
|
|
label=report_kind,
|
|
tenant_id=campaign.tenant_id,
|
|
source="campaigns.report",
|
|
details={
|
|
"resource_type": "campaign_report",
|
|
"campaign_id": campaign.id,
|
|
"campaign_version_id": version.id,
|
|
"report_kind": report_kind,
|
|
"persisted": False,
|
|
"authorization_inherited_from": {
|
|
"resource_type": "campaign",
|
|
"resource_id": campaign.id,
|
|
},
|
|
},
|
|
)
|
|
else:
|
|
return ()
|
|
if campaign is None or (principal.tenant_id and campaign.tenant_id != principal.tenant_id):
|
|
return _missing_resource_provenance(
|
|
principal,
|
|
resource_type=normalized_type,
|
|
resource_id=resource_id,
|
|
)
|
|
items = ([child_item] if child_item is not None else []) + [
|
|
AccessDecisionProvenance(
|
|
kind="resource",
|
|
id=campaign.id,
|
|
label=campaign.name,
|
|
tenant_id=campaign.tenant_id,
|
|
source="campaigns.campaign",
|
|
details={
|
|
"resource_type": "campaign",
|
|
"external_id": campaign.external_id,
|
|
"status": campaign.status,
|
|
},
|
|
)
|
|
]
|
|
items.extend(_owner_provenance(campaign, principal))
|
|
items.extend(_tenant_admin_provenance(principal))
|
|
permission_values = (
|
|
{"read", "write"}
|
|
if action in READ_ACTIONS or action.strip().lower().endswith(":read")
|
|
else {"write"}
|
|
)
|
|
shares = (
|
|
session.query(CampaignShare) # type: ignore[attr-defined]
|
|
.filter(
|
|
CampaignShare.tenant_id == campaign.tenant_id,
|
|
CampaignShare.campaign_id == campaign.id,
|
|
CampaignShare.revoked_at.is_(None),
|
|
CampaignShare.permission.in_(sorted(permission_values)),
|
|
or_(
|
|
(CampaignShare.target_type == "user") & (CampaignShare.target_id == principal.membership_id),
|
|
(CampaignShare.target_type == "group") & (CampaignShare.target_id.in_(sorted(principal.group_ids))),
|
|
),
|
|
)
|
|
.order_by(CampaignShare.target_type.asc(), CampaignShare.target_id.asc())
|
|
.all()
|
|
)
|
|
for share in shares:
|
|
items.append(
|
|
AccessDecisionProvenance(
|
|
kind="share",
|
|
id=share.id,
|
|
label=share.permission,
|
|
tenant_id=share.tenant_id,
|
|
source="campaigns.share",
|
|
details={
|
|
"target_type": share.target_type,
|
|
"target_id": share.target_id,
|
|
"permission": share.permission,
|
|
},
|
|
)
|
|
)
|
|
return tuple(items)
|
|
|
|
|
|
def access_capability(context: object) -> CampaignAccessService:
|
|
return CampaignAccessService()
|
|
|
|
|
|
class CampaignOwnershipService:
|
|
def current_owner(
|
|
self,
|
|
session: object,
|
|
*,
|
|
tenant_id: str,
|
|
resource_id: str,
|
|
) -> OwnershipSubjectRef | None:
|
|
campaign = session.get(Campaign, resource_id) # type: ignore[attr-defined]
|
|
if campaign is None or campaign.tenant_id != tenant_id:
|
|
return None
|
|
if campaign.owner_user_id:
|
|
return OwnershipSubjectRef(type="user", id=campaign.owner_user_id)
|
|
if campaign.owner_group_id:
|
|
return OwnershipSubjectRef(type="group", id=campaign.owner_group_id)
|
|
return None
|
|
|
|
def authorize_ownership_action(
|
|
self,
|
|
session: object,
|
|
*,
|
|
tenant_id: str,
|
|
resource_id: str,
|
|
action: str,
|
|
actor: OwnershipSubjectRef,
|
|
current_owner: OwnershipSubjectRef,
|
|
target_owner: OwnershipSubjectRef,
|
|
) -> OwnershipActionDecision:
|
|
campaign = session.get(Campaign, resource_id) # type: ignore[attr-defined]
|
|
if campaign is None or campaign.tenant_id != tenant_id:
|
|
return OwnershipActionDecision(False, "Campaign was not found")
|
|
if target_owner.type not in {"user", "group"}:
|
|
return OwnershipActionDecision(
|
|
False,
|
|
"Campaign ownership can be assigned only to a user or group",
|
|
)
|
|
|
|
actor_is_current_owner = _actor_controls_owner(actor, current_owner)
|
|
actor_is_target = _actor_controls_owner(actor, target_owner)
|
|
actor_can_share = scopes_grant_compatible(
|
|
actor.scopes,
|
|
"campaigns:campaign:share",
|
|
)
|
|
actor_can_accept_for_group = scopes_grant_compatible(
|
|
actor.scopes,
|
|
"campaigns:ownership:accept_group",
|
|
)
|
|
actor_can_recover = scopes_grant_compatible(
|
|
actor.scopes,
|
|
"campaigns:ownership:recover",
|
|
)
|
|
actor_can_read = (
|
|
scopes_grant_compatible(
|
|
actor.scopes,
|
|
"campaigns:campaign:read",
|
|
)
|
|
and CampaignAccessService().can_read_campaign(
|
|
session,
|
|
tenant_id=tenant_id,
|
|
campaign_id=resource_id,
|
|
user_id=actor.id,
|
|
group_ids=actor.group_ids,
|
|
tenant_admin=scopes_grant_compatible(
|
|
actor.scopes,
|
|
"tenant:*",
|
|
),
|
|
)
|
|
)
|
|
|
|
if action in {
|
|
"propose_transfer",
|
|
"request_ownership",
|
|
"request_recovery",
|
|
}:
|
|
target_decision = _valid_campaign_owner_target(
|
|
tenant_id=tenant_id,
|
|
target_owner=target_owner,
|
|
)
|
|
if target_decision is not None:
|
|
return target_decision
|
|
|
|
if action == "view_transfer":
|
|
allowed = (
|
|
_actor_is_owner_member(actor, current_owner)
|
|
or _actor_is_owner_member(actor, target_owner)
|
|
or actor_can_recover
|
|
)
|
|
elif action in {
|
|
"propose_transfer",
|
|
"approve_requested_transfer",
|
|
}:
|
|
allowed = actor_is_current_owner and actor_can_share
|
|
elif action == "cancel_transfer":
|
|
allowed = (
|
|
(actor_is_current_owner and actor_can_share)
|
|
or actor_is_target
|
|
)
|
|
elif action == "request_ownership":
|
|
allowed = actor_can_read and actor_is_target and (
|
|
target_owner.type != "group" or actor_can_accept_for_group
|
|
)
|
|
elif action == "accept_transfer":
|
|
allowed = target_owner.type == "user" and actor_is_target
|
|
elif action == "accept_group_transfer":
|
|
allowed = (
|
|
target_owner.type == "group"
|
|
and actor_is_target
|
|
and actor_can_accept_for_group
|
|
)
|
|
elif action == "decline_transfer":
|
|
allowed = (
|
|
actor_is_target
|
|
and (
|
|
target_owner.type != "group"
|
|
or actor_can_accept_for_group
|
|
)
|
|
) or (
|
|
actor_is_current_owner
|
|
and actor_can_share
|
|
)
|
|
elif action in {
|
|
"request_recovery",
|
|
"approve_recovery",
|
|
"execute_recovery",
|
|
}:
|
|
allowed = actor_can_recover
|
|
else:
|
|
allowed = False
|
|
|
|
return OwnershipActionDecision(
|
|
allowed=allowed,
|
|
reason=None if allowed else f"Campaign ownership action is not allowed: {action}",
|
|
)
|
|
|
|
def apply_owner(
|
|
self,
|
|
session: object,
|
|
*,
|
|
tenant_id: str,
|
|
resource_id: str,
|
|
expected_owner: OwnershipSubjectRef,
|
|
target_owner: OwnershipSubjectRef,
|
|
actor: OwnershipSubjectRef,
|
|
reason: str | None,
|
|
) -> None:
|
|
del actor, reason
|
|
campaign = session.get(Campaign, resource_id) # type: ignore[attr-defined]
|
|
if campaign is None or campaign.tenant_id != tenant_id:
|
|
raise OwnershipTransferError("Campaign was not found")
|
|
if target_owner.type not in {"user", "group"}:
|
|
raise OwnershipTransferError(
|
|
"Campaign ownership can be assigned only to a user or group"
|
|
)
|
|
from govoplan_campaign.backend.route_support import (
|
|
_clear_current_version_mail_profile_for_owner_transfer,
|
|
)
|
|
|
|
try:
|
|
_clear_current_version_mail_profile_for_owner_transfer(
|
|
session, # type: ignore[arg-type]
|
|
campaign,
|
|
)
|
|
except HTTPException as exc:
|
|
raise OwnershipTransferError(
|
|
str(
|
|
exc.detail
|
|
or "Campaign owner cannot be changed in its current state"
|
|
)
|
|
) from exc
|
|
expected_filter = (
|
|
(
|
|
Campaign.owner_user_id == expected_owner.id,
|
|
Campaign.owner_group_id.is_(None),
|
|
)
|
|
if expected_owner.type == "user"
|
|
else (
|
|
Campaign.owner_user_id.is_(None),
|
|
Campaign.owner_group_id == expected_owner.id,
|
|
)
|
|
)
|
|
updated = (
|
|
session.query(Campaign) # type: ignore[attr-defined]
|
|
.filter(
|
|
Campaign.id == resource_id,
|
|
Campaign.tenant_id == tenant_id,
|
|
*expected_filter,
|
|
)
|
|
.update(
|
|
{
|
|
Campaign.owner_user_id: (
|
|
target_owner.id
|
|
if target_owner.type == "user"
|
|
else None
|
|
),
|
|
Campaign.owner_group_id: (
|
|
target_owner.id
|
|
if target_owner.type == "group"
|
|
else None
|
|
),
|
|
},
|
|
synchronize_session=False,
|
|
)
|
|
)
|
|
if updated != 1:
|
|
raise OwnershipTransferError(
|
|
"Campaign owner changed while the transfer was pending"
|
|
)
|
|
session.expire(campaign) # type: ignore[attr-defined]
|
|
|
|
|
|
def _valid_campaign_owner_target(
|
|
*,
|
|
tenant_id: str,
|
|
target_owner: OwnershipSubjectRef,
|
|
) -> OwnershipActionDecision | None:
|
|
from govoplan_campaign.backend.route_support import _access_directory
|
|
|
|
directory = _access_directory()
|
|
if target_owner.type == "user":
|
|
item = directory.get_user(target_owner.id)
|
|
elif target_owner.type == "group":
|
|
item = directory.get_group(target_owner.id)
|
|
else:
|
|
return OwnershipActionDecision(
|
|
False,
|
|
"Campaign ownership can be assigned only to a user or group",
|
|
)
|
|
if (
|
|
item is None
|
|
or item.tenant_id != tenant_id
|
|
or item.status != "active"
|
|
):
|
|
return OwnershipActionDecision(
|
|
False,
|
|
f"Target {target_owner.type} is not active in this tenant",
|
|
)
|
|
return None
|
|
|
|
|
|
def _actor_is_owner_member(
|
|
actor: OwnershipSubjectRef,
|
|
owner: OwnershipSubjectRef,
|
|
) -> bool:
|
|
if owner.type == "group":
|
|
return owner.id in actor.group_ids
|
|
return actor.type == owner.type and actor.id == owner.id
|
|
|
|
|
|
def _actor_controls_owner(
|
|
actor: OwnershipSubjectRef,
|
|
owner: OwnershipSubjectRef,
|
|
) -> bool:
|
|
return _actor_is_owner_member(actor, owner)
|
|
|
|
|
|
def _missing_resource_provenance(
|
|
principal: PrincipalRef,
|
|
*,
|
|
resource_type: str,
|
|
resource_id: str,
|
|
reason: str = "not_found",
|
|
) -> tuple[AccessDecisionProvenance, ...]:
|
|
return (
|
|
AccessDecisionProvenance(
|
|
kind="resource",
|
|
id=resource_id,
|
|
tenant_id=principal.tenant_id,
|
|
source="campaigns.not_found",
|
|
details={
|
|
"resource_type": resource_type,
|
|
"found": False,
|
|
"reason": reason,
|
|
},
|
|
),
|
|
)
|
|
|
|
|
|
def _owner_provenance(campaign: Campaign, principal: PrincipalRef) -> tuple[AccessDecisionProvenance, ...]:
|
|
if campaign.owner_user_id and campaign.owner_user_id == principal.membership_id:
|
|
return (
|
|
AccessDecisionProvenance(
|
|
kind="owner",
|
|
id=campaign.owner_user_id,
|
|
tenant_id=campaign.tenant_id,
|
|
source="campaigns.owner",
|
|
details={"owner_type": "user"},
|
|
),
|
|
)
|
|
if campaign.owner_group_id and campaign.owner_group_id in principal.group_ids:
|
|
return (
|
|
AccessDecisionProvenance(
|
|
kind="owner",
|
|
id=campaign.owner_group_id,
|
|
tenant_id=campaign.tenant_id,
|
|
source="campaigns.owner",
|
|
details={"owner_type": "group"},
|
|
),
|
|
)
|
|
return ()
|
|
|
|
|
|
def _tenant_admin_provenance(principal: PrincipalRef) -> tuple[AccessDecisionProvenance, ...]:
|
|
if not scopes_grant_compatible(principal.scopes, "tenant:*"):
|
|
return ()
|
|
return (
|
|
AccessDecisionProvenance(
|
|
kind="policy",
|
|
id="tenant:*",
|
|
label="tenant:*",
|
|
tenant_id=principal.tenant_id,
|
|
source="campaigns.tenant_admin",
|
|
details={"grant": "tenant_admin"},
|
|
),
|
|
)
|
|
|
|
|
|
class CampaignPolicyContextService(CampaignPolicyContextProvider):
|
|
def get_campaign_policy_context(
|
|
self,
|
|
session: object,
|
|
*,
|
|
tenant_id: str | None = None,
|
|
campaign_id: str,
|
|
) -> CampaignPolicyContext | None:
|
|
campaign = session.get(Campaign, campaign_id) # type: ignore[attr-defined]
|
|
if campaign is None or (tenant_id is not None and campaign.tenant_id != tenant_id):
|
|
return None
|
|
return CampaignPolicyContext(
|
|
id=campaign.id,
|
|
tenant_id=campaign.tenant_id,
|
|
owner_user_id=campaign.owner_user_id,
|
|
owner_group_id=campaign.owner_group_id,
|
|
settings=dict(campaign.settings or {}),
|
|
)
|
|
|
|
def set_campaign_settings(
|
|
self,
|
|
session: object,
|
|
*,
|
|
tenant_id: str,
|
|
campaign_id: str,
|
|
settings: Mapping[str, object],
|
|
) -> Mapping[str, object]:
|
|
campaign = session.get(Campaign, campaign_id) # type: ignore[attr-defined]
|
|
if campaign is None or campaign.tenant_id != tenant_id:
|
|
raise ValueError("Campaign privacy policy not found")
|
|
payload = dict(settings)
|
|
campaign.settings = payload
|
|
session.add(campaign) # type: ignore[attr-defined]
|
|
return payload
|
|
|
|
|
|
def policy_context_capability(context: object) -> CampaignPolicyContextService:
|
|
return CampaignPolicyContextService()
|
|
|
|
|
|
class CampaignDeliveryTaskService(CampaignDeliveryTaskProvider):
|
|
def send_campaign_job(self, session: object, *, job_id: str, enqueue_imap_task: bool = True) -> Mapping[str, object]:
|
|
from govoplan_campaign.backend.sending.jobs import send_campaign_job
|
|
|
|
return send_campaign_job(session, job_id=job_id, enqueue_imap_task=enqueue_imap_task).as_dict() # type: ignore[arg-type]
|
|
|
|
def append_sent_for_job(self, session: object, *, job_id: str) -> Mapping[str, object]:
|
|
from govoplan_campaign.backend.sending.jobs import append_sent_for_job
|
|
|
|
return append_sent_for_job(session, job_id=job_id).as_dict() # type: ignore[arg-type]
|
|
|
|
|
|
def delivery_tasks_capability(context: object) -> CampaignDeliveryTaskService:
|
|
from govoplan_campaign.backend.runtime import configure_runtime
|
|
|
|
configure_runtime(
|
|
registry=getattr(context, "registry", None),
|
|
settings=getattr(context, "settings", None),
|
|
)
|
|
return CampaignDeliveryTaskService()
|
|
|
|
|
|
class CampaignRetentionService(CampaignRetentionProvider):
|
|
def apply_retention(self, session: object, *, dry_run, now, policy_for_campaign_id):
|
|
from govoplan_campaign.backend.retention import apply_campaign_retention
|
|
|
|
return apply_campaign_retention(session, dry_run=dry_run, now=now, policy_for_campaign_id=policy_for_campaign_id) # type: ignore[arg-type]
|
|
|
|
|
|
def retention_capability(context: object) -> CampaignRetentionService:
|
|
return CampaignRetentionService()
|