Release v0.1.5

src/govoplan_campaign/backend/capabilities.py

@@ -0,0 +1,177 @@
+from __future__ import annotations
+
+from collections.abc import Iterable, Mapping
+
+from sqlalchemy import and_, or_
+
+from govoplan_core.core.campaigns import (
+    CampaignAccessProvider,
+    CampaignDeliveryTaskProvider,
+    CampaignMailPolicyContext,
+    CampaignMailPolicyContextProvider,
+    CampaignPolicyContext,
+    CampaignPolicyContextProvider,
+    CampaignRetentionProvider,
+)
+
+from govoplan_campaign.backend.db.models import Campaign, CampaignShare
+
+
+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 access_capability(context: object) -> CampaignAccessService:
+    return CampaignAccessService()
+
+
+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:
+    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()