from __future__ import annotations from pathlib import Path from govoplan_core.core.access import ( CAPABILITY_AUTH_PERMISSION_EVALUATOR, CAPABILITY_AUTH_PRINCIPAL_RESOLVER, ) from govoplan_core.core.module_guards import ( drop_table_retirement_provider, persistent_table_uninstall_guard, ) from govoplan_core.core.datasources import CAPABILITY_DATASOURCE_ORIGINS from govoplan_core.core.feeds import CAPABILITY_CONNECTORS_FEEDS from govoplan_core.core.modules import ( DocumentationTopic, MigrationSpec, ModuleInterfaceProvider, ModuleManifest, PermissionDefinition, RoleTemplate, ) from govoplan_core.core.provider_governance import ( ExternalProviderDeclaration, ExternalProviderStateProviderRegistration, ModuleArchitectureDeclaration, ModuleArchitectureDocumentation, ModuleMaturityEvidence, ProviderBehaviorDeclaration, ProviderObjectDeclaration, ) from govoplan_core.core.tabular_sources import ( CAPABILITY_CONNECTORS_TABULAR_SNAPSHOT_WRITER, CAPABILITY_CONNECTORS_TABULAR_SOURCES, ) from govoplan_core.core.sanctions import ( CAPABILITY_CONNECTORS_SANCTIONS_SNAPSHOTS, ) from govoplan_core.db.base import Base from govoplan_connectors.backend.db.models import ( ConnectorSanctionsAcquisitionRun, ConnectorSanctionsSnapshot, ConnectorTabularSource, ) from govoplan_connectors.backend.sanctions_sources import ( SANCTIONS_READ_SCOPE, SANCTIONS_REFRESH_SCOPE, SqlSanctionsSnapshotProvider, ) from govoplan_connectors.backend.tabular_sources import ( ADMIN_SCOPE, READ_SCOPE, WRITE_SCOPE, SqlTabularSourceProvider, ) from govoplan_connectors.backend.datasource_origins import ( ConnectorDatasourceOriginProvider, ) from govoplan_connectors.backend.feeds import ConnectorFeedProvider from govoplan_connectors.backend.provider_state import ( SANCTIONS_PROVIDER_ID, TABULAR_PROVIDER_ID, sanctions_provider_states, tabular_provider_states, ) MODULE_ID = "connectors" MODULE_VERSION = "0.1.18" TABULAR_SOURCE_INTERFACE_VERSION = "0.1.0" DATASOURCE_ORIGIN_INTERFACE_VERSION = "0.1.0" SANCTIONS_SNAPSHOT_INTERFACE_VERSION = "1.0.0" FEED_INTERFACE_VERSION = "0.1.0" CONNECTOR_RUNTIME_INTERFACE_VERSION = "1.0.0" ARCHITECTURE = ModuleArchitectureDeclaration( layer="data_reporting_integration", kind="integration", maturity="vertical_slice", evidence=( ModuleMaturityEvidence( kind="test", reference="tests/test_tabular_sources.py", summary="Exercises tenant-safe immutable tabular snapshots and bounded reads.", ), ModuleMaturityEvidence( kind="test", reference="tests/test_sanctions_sources.py", summary="Exercises source acquisition health, checksums, retries, and immutable evidence.", ), ModuleMaturityEvidence( kind="recovery", reference="tests/test_recovery.py", summary="Proves atomic snapshot commits, idempotent replay, distributed fences, tamper rejection, and unknown external-effect handling.", ), ModuleMaturityEvidence( kind="documentation", reference="docs/CONNECTOR_SOURCE_LIFECYCLE.md", summary="Defines source lifecycle, authority, evidence, and outage boundaries.", ), ), known_limits=( "The executable generic datasource origin is an immutable tabular snapshot; database and arbitrary REST profiles remain future providers.", "Feed publication renders a governed document but does not yet push it to an external publishing endpoint.", ), supported_authority_modes=( "external_authoritative", "external_mirror", "linked_reference", ), owned_concepts=( "external transport profiles", "protocol interaction", "immutable connector snapshots", "connector acquisition health", ), non_owned_concepts=( "datasource catalogue identity and lifecycle", "domain records and business semantics", "data transformations", "screening dispositions", ), target_tested_providers=( TABULAR_PROVIDER_ID, SANCTIONS_PROVIDER_ID, ), documentation=ModuleArchitectureDocumentation( migration=("src/govoplan_connectors/backend/migrations/versions",), upgrade=("docs/CONNECTOR_SOURCE_LIFECYCLE.md",), recovery=("docs/CONNECTOR_SOURCE_LIFECYCLE.md",), security=("docs/CONNECTOR_SOURCE_LIFECYCLE.md",), operations=("docs/CONNECTOR_SOURCE_LIFECYCLE.md",), ), ) EXTERNAL_PROVIDERS = ( ExternalProviderDeclaration( id=TABULAR_PROVIDER_ID, module_id=MODULE_ID, label="Immutable tabular snapshot provider", maturity="read", operations=("discover", "search", "read", "preview", "dry_run"), objects=( ProviderObjectDeclaration( object_type="tabular_source_snapshot", field_groups=("identity", "schema", "rows", "source_provenance"), authority_modes=("external_authoritative", "external_mirror"), default_authority_mode="external_mirror", ), ), behavior=ProviderBehaviorDeclaration( revision_tokens="Source fingerprints and immutable snapshot ids are retained.", concurrency="Reads may require the expected fingerprint; snapshots never mutate in place.", freshness="Snapshot acquisition time and source timestamp are exposed.", health="Import validation and source-read failures are explicit.", max_read_items=1000, idempotency="Feed imports accept a caller request key and replay the same committed immutable source without refetching.", retry="Read-only acquisition may be retried only as a new deliberate request after a failed atomic operation.", outcome_unknown="Provider reads do not mutate remote state; an uncertain database commit is resolved by the atomic recovery transaction.", outcome_unknown_supported=False, evidence="Rows, schema, fingerprint, source metadata, and acquisition provenance remain linked.", correction="Import a replacement snapshot; retain the prior snapshot as evidence.", rollback="Snapshot rows and the terminal recovery checkpoint commit or roll back together.", reconciliation="Compare source and snapshot fingerprints before selecting a new current state.", outage="Existing snapshots remain available and visibly stale; no live-source claim is made.", classifications=("internal", "confidential", "restricted"), purposes=("governed import", "dataflow input", "evidence reconstruction"), retention="Datasources or the consuming domain supplies retention and hold policy.", secret_handling="Generic snapshots contain no connector credential; transport credentials stay in credential envelopes.", ), capability_names=( CAPABILITY_CONNECTORS_TABULAR_SOURCES, CAPABILITY_DATASOURCE_ORIGINS, ), interface_names=( "connectors.tabular_sources", "connectors.datasource_origins", ), documentation_topic_ids=( "connectors.authority-and-effects", "connectors.tabular-sources", ), ), ExternalProviderDeclaration( id=SANCTIONS_PROVIDER_ID, module_id=MODULE_ID, label="Sanctions source snapshot provider", maturity="read", operations=("discover", "search", "read", "preview"), objects=( ProviderObjectDeclaration( object_type="sanctions_source_snapshot", field_groups=("source_identity", "raw_evidence", "entries", "acquisition_health"), authority_modes=("external_authoritative", "external_mirror"), default_authority_mode="external_mirror", ), ), behavior=ProviderBehaviorDeclaration( revision_tokens="Provider source version, ETag, Last-Modified, and SHA-256 digest are retained when available.", concurrency="Refreshes use conditional source requests, a distributed per-tenant/provider fence, and immutable snapshots.", freshness="Latest successful acquisition, source timestamp, and stale health are reported.", health="Transport, parsing, source-change, and malformed-source states are explicit.", max_read_items=5000, idempotency="A caller request key identifies one acquisition run and replays its committed result without contacting the source again.", retry="Bounded HTTP retries are safe because acquisition is read-only; failed runs require a new deliberate request key.", timeout_seconds=30, outcome_unknown="The external operation is read-only; snapshot rows and recovery evidence commit atomically.", outcome_unknown_supported=False, evidence="Raw source bytes, checksum, acquisition run, parser result, and normalized entry count are linked.", correction="A corrected source creates a new immutable snapshot and acquisition run.", rollback="A failed database transaction leaves no snapshot and the stale atomic fence resolves as failed.", reconciliation="Compare source version and digest, then preserve both prior and corrected evidence.", outage="The latest accepted snapshot stays usable with stale/unavailable source health.", classifications=("public", "internal"), purposes=("sanctions source acquisition", "compliance screening evidence"), retention="Risk and Records policies determine accepted snapshot retention and legal holds.", secret_handling="Public sources require no subject data or source credential; configured proxy secrets remain external to snapshots.", ), capability_names=(CAPABILITY_CONNECTORS_SANCTIONS_SNAPSHOTS,), interface_names=("connectors.sanctions_snapshots",), documentation_topic_ids=( "connectors.authority-and-effects", "connectors.sanctions-snapshots", ), ), ) def _permission(scope: str, label: str, description: str) -> PermissionDefinition: module_id, resource, action = scope.split(":", 2) return PermissionDefinition( scope=scope, label=label, description=description, category="Connectors", level="tenant", module_id=module_id, resource=resource, action=action, ) PERMISSIONS = ( _permission( READ_SCOPE, "View tabular sources", "Discover and preview policy-visible tabular connector sources.", ), _permission( WRITE_SCOPE, "Manage tabular sources", "Import and retire bounded tabular snapshots.", ), _permission( ADMIN_SCOPE, "Administer connector sources", "Manage every tenant connector source and future source policies.", ), _permission( SANCTIONS_READ_SCOPE, "View sanctions source evidence", "Inspect immutable sanctions snapshots and acquisition health.", ), _permission( SANCTIONS_REFRESH_SCOPE, "Refresh sanctions sources", "Acquire a new immutable sanctions source snapshot.", ), ) ROLE_TEMPLATES = ( RoleTemplate( slug="connector_source_manager", name="Connector source manager", description="Discover, import, preview, and retire tabular sources.", permissions=( READ_SCOPE, WRITE_SCOPE, SANCTIONS_READ_SCOPE, SANCTIONS_REFRESH_SCOPE, ), ), RoleTemplate( slug="connector_source_reader", name="Connector source reader", description="Discover and preview tabular connector sources.", permissions=(READ_SCOPE, SANCTIONS_READ_SCOPE), ), ) def _router(_context): from govoplan_connectors.backend.router import router return router def _provider(_context) -> SqlTabularSourceProvider: return SqlTabularSourceProvider() def _datasource_origin_provider(_context) -> ConnectorDatasourceOriginProvider: return ConnectorDatasourceOriginProvider() def _sanctions_snapshot_provider( _context, ) -> SqlSanctionsSnapshotProvider: return SqlSanctionsSnapshotProvider() def _feed_provider(_context) -> ConnectorFeedProvider: return ConnectorFeedProvider() def _tenant_summary(session, tenant_id: str) -> dict[str, int]: return { "connector_tabular_sources": ( session.query(ConnectorTabularSource) .filter( ConnectorTabularSource.tenant_id == tenant_id, ConnectorTabularSource.deleted_at.is_(None), ) .count() ), "connector_sanctions_snapshots": ( session.query(ConnectorSanctionsSnapshot) .filter(ConnectorSanctionsSnapshot.tenant_id == tenant_id) .count() ), "connector_sanctions_runs": ( session.query(ConnectorSanctionsAcquisitionRun) .filter(ConnectorSanctionsAcquisitionRun.tenant_id == tenant_id) .count() ), } manifest = ModuleManifest( id=MODULE_ID, name="Connectors", version=MODULE_VERSION, optional_dependencies=( "access", "audit", "files", "policy", "datasources", "portal", "reporting", "risk_compliance", ), required_capabilities=( CAPABILITY_AUTH_PRINCIPAL_RESOLVER, CAPABILITY_AUTH_PERMISSION_EVALUATOR, ), provides_interfaces=( ModuleInterfaceProvider( name="connectors.tabular_sources", version=TABULAR_SOURCE_INTERFACE_VERSION, ), ModuleInterfaceProvider( name="connectors.tabular_snapshot_writer", version=TABULAR_SOURCE_INTERFACE_VERSION, ), ModuleInterfaceProvider( name="connectors.datasource_origins", version=DATASOURCE_ORIGIN_INTERFACE_VERSION, ), ModuleInterfaceProvider( name="connectors.sanctions_snapshots", version=SANCTIONS_SNAPSHOT_INTERFACE_VERSION, ), ModuleInterfaceProvider( name="connectors.feeds", version=FEED_INTERFACE_VERSION, ), ModuleInterfaceProvider( name="connectors.runtime_contract", version=CONNECTOR_RUNTIME_INTERFACE_VERSION, ), ), permissions=PERMISSIONS, role_templates=ROLE_TEMPLATES, route_factory=_router, capability_factories={ CAPABILITY_CONNECTORS_TABULAR_SOURCES: _provider, CAPABILITY_CONNECTORS_TABULAR_SNAPSHOT_WRITER: _provider, CAPABILITY_DATASOURCE_ORIGINS: _datasource_origin_provider, CAPABILITY_CONNECTORS_SANCTIONS_SNAPSHOTS: (_sanctions_snapshot_provider), CAPABILITY_CONNECTORS_FEEDS: _feed_provider, }, tenant_summary_providers=(_tenant_summary,), architecture=ARCHITECTURE, external_providers=EXTERNAL_PROVIDERS, external_provider_state_providers=( ExternalProviderStateProviderRegistration( module_id=MODULE_ID, provider_id=TABULAR_PROVIDER_ID, provider=tabular_provider_states, ), ExternalProviderStateProviderRegistration( module_id=MODULE_ID, provider_id=SANCTIONS_PROVIDER_ID, provider=sanctions_provider_states, ), ), migration_spec=MigrationSpec( module_id=MODULE_ID, metadata=Base.metadata, script_location=str(Path(__file__).with_name("migrations") / "versions"), retirement_supported=True, retirement_provider=drop_table_retirement_provider( ConnectorSanctionsSnapshot, ConnectorSanctionsAcquisitionRun, ConnectorTabularSource, label="Connectors", ), retirement_notes=( "Destructive retirement drops connector-owned source snapshots after " "the installer captures a database snapshot." ), ), uninstall_guard_providers=( persistent_table_uninstall_guard( ConnectorSanctionsSnapshot, ConnectorSanctionsAcquisitionRun, ConnectorTabularSource, label="Connectors", ), ), documentation=( DocumentationTopic( id="connectors.authority-and-effects", title="Connector authority and effect behavior", summary="Connector direction, technical maturity, and configured source authority are separate and must remain visible.", body=( "A connector can consume, publish, or work bidirectionally and can mature from discovery through replacement. " "Each binding separately states whether GovOPlaN is authoritative, follows an external authority, keeps a mirror, synchronizes under conflict rules, adds a governance overlay, or retains only a link. " "Writable providers must explain revisions, limits, idempotency, outcome-unknown handling, evidence, reconciliation, correction, outage behavior, and secret requirements." ), layer="available", documentation_types=("admin", "user"), audience=("operator", "module_admin", "power_user", "product_owner"), related_modules=("datasources", "dataflow", "ops", "policy", "audit"), order=39, ), DocumentationTopic( id="connectors.runtime-preview-contract", title="Connector previews and diagnostics", summary="Use one bounded, redacted dry-run shape across external transports.", body=( "Connectors owns endpoint discovery, authentication hand-off, transport limits, retries, and protocol health. " "Domain modules own field mapping, validation, reconciliation, and record mutation. The shared Core runtime " "contract reports redacted effects and diagnostics with source revisions, fingerprints, and immutable input hashes. " "Tabular previews enforce effective row, serialized-byte, and elapsed-time ceilings and report limit truncation " "as structured diagnostics. A commit must reject stale, truncated, conflicting, or error-bearing previews, and " "credentials never appear in URLs or samples." ), layer="available", documentation_types=("admin", "user"), audience=("operator", "module_admin", "power_user"), related_modules=("addresses", "datasources", "dataflow", "policy", "audit"), order=40, ), DocumentationTopic( id="connectors.tabular-sources", title="Governed tabular sources", summary="Provider-neutral source discovery and bounded reads for Dataflow.", body=( "Connectors owns source configuration, access checks, schema discovery, " "fingerprints, and bounded reads. Dataflow stores only opaque source " "references and expected fingerprints. Each source declares its live, " "cached, file-backed, or static mode, structured health, and supported " "projection, filter, aggregation, sorting, and pagination pushdown. The " "first executable provider imports immutable JSON or CSV snapshots, " "supports projection and pagination, and exposes them as Datasource " "origins. Database and API providers can implement the same origin " "contract without changing Datasources or Dataflow." ), layer="available", documentation_types=("admin", "user"), audience=("operator", "module_admin", "power_user"), related_modules=("dataflow", "files", "reporting", "risk_compliance"), order=40, ), DocumentationTopic( id="connectors.rss-atom", title="RSS and Atom feeds", summary="Import governed feed snapshots and emit visibility-filtered feeds.", body=( "Connectors owns bounded, SSRF-protected RSS/Atom transport and XML " "parsing. Imported entries become immutable tabular snapshots exposed " "through Datasources, including acquisition, freshness, ETag, content " "digest, and source provenance. Portal or Reporting owns publication " "routes and must pass the allowed visibility set when rendering output. " "A separate RSS module is only warranted if GovOPlaN later needs a " "dedicated feed-reader product surface." ), layer="available", documentation_types=("admin", "user"), audience=("operator", "module_admin", "power_user"), related_modules=("datasources", "dataflow", "portal", "reporting"), order=42, ), DocumentationTopic( id="connectors.sanctions-snapshots", title="Sanctions source snapshots", summary=( "Acquire immutable, checksum-verifiable sanctions list " "evidence without transmitting screening subjects." ), body=( "Connectors provides a deterministic synthetic fixture and " "the official United Nations Security Council consolidated " "XML source. Each fetch records conditional transport " "evidence, bounded retries, health state, source metadata, " "raw evidence, and a SHA-256 checksum. Refreshes acquire a " "distributed recovery fence before provider I/O; the immutable " "snapshot and terminal recovery checkpoint then commit in one " "transaction. A repeated request key returns the same result. " "Risk Compliance owns " "normalization, matching, legal review, and dispositions." ), layer="available", documentation_types=("admin", "user"), audience=("operator", "module_admin", "compliance_reviewer"), related_modules=("risk_compliance", "dataflow"), order=41, ), ), ) def get_manifest() -> ModuleManifest: return manifest __all__ = [ "MODULE_ID", "MODULE_VERSION", "DATASOURCE_ORIGIN_INTERFACE_VERSION", "SANCTIONS_SNAPSHOT_INTERFACE_VERSION", "TABULAR_SOURCE_INTERFACE_VERSION", "get_manifest", "manifest", ]