feat: expose sanctions screening integration

src/govoplan_core/core/sanctions.py

@@ -0,0 +1,152 @@
+from __future__ import annotations
+
+from collections.abc import Mapping, Sequence
+from dataclasses import dataclass, field
+from datetime import datetime
+from typing import Protocol, runtime_checkable
+
+
+SANCTIONS_SNAPSHOT_CONTRACT_VERSION = "1"
+CAPABILITY_CONNECTORS_SANCTIONS_SNAPSHOTS = (
+    "connectors.sanctionsSnapshotProvider"
+)
+
+
+@dataclass(frozen=True, slots=True)
+class SanctionsSnapshotReference:
+    ref: str
+    tenant_id: str
+    provider_id: str
+    publisher: str
+    jurisdiction: str
+    list_type: str
+    source_id: str
+    source_version: str
+    publication_at: datetime | None
+    effective_at: datetime | None
+    acquired_at: datetime
+    content_type: str
+    byte_count: int
+    sha256: str
+    parser_version: str
+    raw_evidence_ref: str
+    connector_run_id: str
+    signature_evidence: Mapping[str, object] = field(
+        default_factory=dict
+    )
+    licence_notes: str | None = None
+    trust_notes: str | None = None
+    transport_evidence: Mapping[str, object] = field(
+        default_factory=dict
+    )
+    contract_version: str = SANCTIONS_SNAPSHOT_CONTRACT_VERSION
+
+    def __post_init__(self) -> None:
+        if self.contract_version != SANCTIONS_SNAPSHOT_CONTRACT_VERSION:
+            raise ValueError(
+                "Unsupported sanctions snapshot contract version."
+            )
+        required = (
+            self.ref,
+            self.tenant_id,
+            self.provider_id,
+            self.publisher,
+            self.jurisdiction,
+            self.list_type,
+            self.source_id,
+            self.source_version,
+            self.content_type,
+            self.sha256,
+            self.parser_version,
+            self.raw_evidence_ref,
+            self.connector_run_id,
+        )
+        if not all(value.strip() for value in required):
+            raise ValueError(
+                "Sanctions snapshot references require complete provenance."
+            )
+        if self.byte_count < 1:
+            raise ValueError(
+                "Sanctions snapshot evidence must not be empty."
+            )
+        if (
+            len(self.sha256) != 64
+            or any(character not in "0123456789abcdef" for character in self.sha256)
+        ):
+            raise ValueError(
+                "Sanctions snapshot SHA-256 evidence is invalid."
+            )
+
+
+@dataclass(frozen=True, slots=True)
+class SanctionsSnapshotPayload:
+    snapshot: SanctionsSnapshotReference
+    content: bytes
+
+    def __post_init__(self) -> None:
+        if len(self.content) != self.snapshot.byte_count:
+            raise ValueError(
+                "Sanctions snapshot content length does not match metadata."
+            )
+
+
+@runtime_checkable
+class SanctionsSnapshotProvider(Protocol):
+    def list_snapshots(
+        self,
+        session: object,
+        principal: object,
+        *,
+        limit: int = 100,
+    ) -> Sequence[SanctionsSnapshotReference]:
+        ...
+
+    def get_snapshot(
+        self,
+        session: object,
+        principal: object,
+        *,
+        snapshot_ref: str,
+    ) -> SanctionsSnapshotReference | None:
+        ...
+
+    def read_snapshot(
+        self,
+        session: object,
+        principal: object,
+        *,
+        snapshot_ref: str,
+    ) -> SanctionsSnapshotPayload:
+        ...
+
+
+def sanctions_snapshot_provider(
+    registry: object | None,
+) -> SanctionsSnapshotProvider | None:
+    if (
+        registry is None
+        or not hasattr(registry, "has_capability")
+        or not hasattr(registry, "capability")
+        or not registry.has_capability(
+            CAPABILITY_CONNECTORS_SANCTIONS_SNAPSHOTS
+        )
+    ):
+        return None
+    provider = registry.capability(
+        CAPABILITY_CONNECTORS_SANCTIONS_SNAPSHOTS
+    )
+    return (
+        provider
+        if isinstance(provider, SanctionsSnapshotProvider)
+        else None
+    )
+
+
+__all__ = [
+    "CAPABILITY_CONNECTORS_SANCTIONS_SNAPSHOTS",
+    "SANCTIONS_SNAPSHOT_CONTRACT_VERSION",
+    "SanctionsSnapshotPayload",
+    "SanctionsSnapshotProvider",
+    "SanctionsSnapshotReference",
+    "sanctions_snapshot_provider",
+]

src/govoplan_core/settings.py

@@ -49,8 +49,8 @@ class Settings(BaseSettings):
         default=(
             "tenancy,organizations,identity,idm,access,admin,dashboard,policy,"
             "audit,campaigns,files,mail,calendar,poll,scheduling,connectors,"
-            "datasources,dataflow,workflow,views,search,postbox,notifications,"
-            "docs,ops"
+            "datasources,dataflow,workflow,views,search,risk_compliance,"
+            "postbox,notifications,docs,ops"
         ),
         alias="ENABLED_MODULES",
     )