feat: expose connector datasource origins

README.md

@@ -15,13 +15,18 @@ or reporting behavior.
 ## Executable First Slice
 
 The first executable connector capability provides tenant-isolated tabular
-sources for Dataflow. Operators can import bounded JSON or CSV snapshots,
-inspect inferred schemas, and expose immutable source references and content
-fingerprints through `connectors.tabular_sources@0.1.0`.
+origins. Operators can import bounded JSON or CSV snapshots, inspect inferred
+schemas, and expose immutable source references and content fingerprints
+through `connectors.datasource_origins@0.1.0`.
 
-Database, API, file, and warehouse providers can implement the same core
-capability later. Dataflow stores opaque references and expected fingerprints;
-it never stores connector credentials.
+Connectors owns acquisition, connection profiles, credentials, discovery, and
+provider health. `govoplan-datasources` registers an origin as a governed live
+or cached datasource and owns staging, materializations, frozen states, and
+consumer access. Dataflow consumes that Datasources contract and never imports
+connector implementations or stores connector credentials.
+
+Database, REST/HTTP, directory, managed-file, and warehouse providers can
+implement the same origin contract without changing Datasources or Dataflow.
 
 Development:
 

docs/CONCEPT.md

@@ -32,6 +32,8 @@ The module owns:
 
 The module does not own:
 
+- governed datasource identity, staging, materializations, frozen states, or
+  consumer read semantics, owned by `govoplan-datasources`
 - file storage semantics, owned by files/DMS
 - identity provisioning semantics, owned by IDM/access
 - mail/calendar semantics, owned by mail/calendar
@@ -69,6 +71,11 @@ Domain modules should ask whether a connector capability exists and request a
 profile/test result through core-mediated capabilities. They must not import
 connector implementation modules directly.
 
+Data-oriented consumers use a two-layer path: Connectors publishes a
+provider-specific datasource origin, then Datasources registers and governs it.
+Dataflow, Workflow, Reporting, and other consumers use Datasources rather than
+calling the connector origin directly.
+
 ## Reference Journeys
 
 ### OpenProject Connector First

docs/CONNECTOR_SOURCE_LIFECYCLE.md

@@ -88,10 +88,12 @@ this lifecycle when a connector publishes status.
 2. Fetch only the minimal remote data required for the declared use case.
 3. Normalize into a connector-owned staging payload.
 4. Validate shape, required fields, and source trust level.
-5. Emit a core-mediated event such as `connector.record_discovered`.
-6. Let domain modules claim or transform staged data through capabilities, not
-   imports.
-7. Store external references with source system, object type, object id, version
+5. Publish data-shaped inputs as versioned datasource origins.
+6. Let Datasources register live/cached origins or stage immutable snapshots.
+7. Let domain modules consume governed datasource references through
+   capabilities, not imports.
+8. Emit a core-mediated event such as `connector.record_discovered`.
+9. Store external references with source system, object type, object id, version
    or ETag, and last-seen timestamp.
 
 ## Publish Flow
@@ -124,6 +126,7 @@ should ask core for capabilities such as:
 - `connectors.profileTester`
 - `connectors.health`
 - `connectors.externalReferences`
+- `connectors.datasourceOrigins`
 - `connectors.sourceConsumer`
 - `connectors.sourcePublisher`
 

src/govoplan_connectors/backend/datasource_origins.py

@@ -0,0 +1,132 @@
+from __future__ import annotations
+
+from govoplan_core.core.datasources import (
+    DatasourceAccessError,
+    DatasourceField,
+    DatasourceNotFoundError,
+    DatasourceOrigin,
+    DatasourceOriginReadRequest,
+    DatasourceOriginReadResult,
+    DatasourceValidationError,
+)
+from govoplan_core.core.tabular_sources import (
+    TabularReadRequest,
+    TabularSource,
+    TabularSourceAccessError,
+    TabularSourceError,
+    TabularSourceNotFoundError,
+    TabularSourceValidationError,
+)
+from govoplan_connectors.backend.tabular_sources import SqlTabularSourceProvider
+
+
+class ConnectorDatasourceOriginProvider:
+    """Expose connector-owned sources through the Datasources origin contract."""
+
+    def __init__(self, provider: SqlTabularSourceProvider | None = None) -> None:
+        self._provider = provider or SqlTabularSourceProvider()
+
+    def list_origins(
+        self,
+        session: object,
+        principal: object,
+        *,
+        query: str = "",
+        limit: int = 100,
+    ):
+        try:
+            rows = self._provider.list_sources(
+                session,
+                principal,
+                query=query,
+                limit=limit,
+            )
+        except TabularSourceError as exc:
+            raise _datasource_error(exc) from exc
+        return tuple(_origin(source) for source in rows)
+
+    def get_origin(
+        self,
+        session: object,
+        principal: object,
+        *,
+        origin_ref: str,
+    ) -> DatasourceOrigin | None:
+        try:
+            source = self._provider.get_source(
+                session,
+                principal,
+                source_ref=origin_ref,
+            )
+        except TabularSourceError as exc:
+            raise _datasource_error(exc) from exc
+        return _origin(source) if source is not None else None
+
+    def read_origin(
+        self,
+        session: object,
+        principal: object,
+        *,
+        request: DatasourceOriginReadRequest,
+    ) -> DatasourceOriginReadResult:
+        try:
+            result = self._provider.read_source(
+                session,
+                principal,
+                request=TabularReadRequest(
+                    source_ref=request.origin_ref,
+                    limit=request.limit,
+                    offset=request.offset,
+                    columns=request.columns,
+                    expected_fingerprint=request.expected_fingerprint,
+                ),
+            )
+        except TabularSourceError as exc:
+            raise _datasource_error(exc) from exc
+        return DatasourceOriginReadResult(
+            origin=_origin(result.source),
+            rows=result.rows,
+            total_rows=result.total_rows,
+            truncated=result.truncated,
+        )
+
+
+def _origin(source: TabularSource) -> DatasourceOrigin:
+    return DatasourceOrigin(
+        ref=source.ref,
+        source_name=source.source_name,
+        name=source.name,
+        description=source.description,
+        kind="upload",
+        shape="tabular",
+        supported_modes=("live", "cached"),
+        provider=f"connectors.{source.provider}",
+        schema=tuple(
+            DatasourceField(
+                name=column.name,
+                data_type=column.data_type,
+                nullable=column.nullable,
+            )
+            for column in source.schema
+        ),
+        schema_version=source.schema_version,
+        fingerprint=source.fingerprint,
+        row_count=source.row_count,
+        byte_count=source.byte_count,
+        updated_at=source.updated_at,
+        capabilities=source.capabilities,
+        metadata=dict(source.metadata),
+    )
+
+
+def _datasource_error(exc: TabularSourceError):
+    if isinstance(exc, TabularSourceAccessError):
+        return DatasourceAccessError(str(exc))
+    if isinstance(exc, TabularSourceNotFoundError):
+        return DatasourceNotFoundError(str(exc))
+    if isinstance(exc, TabularSourceValidationError):
+        return DatasourceValidationError(str(exc))
+    return DatasourceValidationError(str(exc))
+
+
+__all__ = ["ConnectorDatasourceOriginProvider"]

src/govoplan_connectors/backend/manifest.py

@@ -10,6 +10,7 @@ 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.modules import (
     DocumentationTopic,
     MigrationSpec,
@@ -30,11 +31,15 @@ from govoplan_connectors.backend.tabular_sources import (
     WRITE_SCOPE,
     SqlTabularSourceProvider,
 )
+from govoplan_connectors.backend.datasource_origins import (
+    ConnectorDatasourceOriginProvider,
+)
 
 
 MODULE_ID = "connectors"
 MODULE_VERSION = "0.1.14"
 TABULAR_SOURCE_INTERFACE_VERSION = "0.1.0"
+DATASOURCE_ORIGIN_INTERFACE_VERSION = "0.1.0"
 
 
 def _permission(scope: str, label: str, description: str) -> PermissionDefinition:
@@ -95,6 +100,10 @@ def _provider(_context) -> SqlTabularSourceProvider:
     return SqlTabularSourceProvider()
 
 
+def _datasource_origin_provider(_context) -> ConnectorDatasourceOriginProvider:
+    return ConnectorDatasourceOriginProvider()
+
+
 def _tenant_summary(session, tenant_id: str) -> dict[str, int]:
     return {
         "connector_tabular_sources": (
@@ -126,6 +135,10 @@ manifest = ModuleManifest(
             name="connectors.tabular_snapshot_writer",
             version=TABULAR_SOURCE_INTERFACE_VERSION,
         ),
+        ModuleInterfaceProvider(
+            name="connectors.datasource_origins",
+            version=DATASOURCE_ORIGIN_INTERFACE_VERSION,
+        ),
     ),
     permissions=PERMISSIONS,
     role_templates=ROLE_TEMPLATES,
@@ -133,6 +146,7 @@ manifest = ModuleManifest(
     capability_factories={
         CAPABILITY_CONNECTORS_TABULAR_SOURCES: _provider,
         CAPABILITY_CONNECTORS_TABULAR_SNAPSHOT_WRITER: _provider,
+        CAPABILITY_DATASOURCE_ORIGINS: _datasource_origin_provider,
     },
     tenant_summary_providers=(_tenant_summary,),
     migration_spec=MigrationSpec(
@@ -164,8 +178,9 @@ manifest = ModuleManifest(
                 "Connectors owns source configuration, access checks, schema discovery, "
                 "fingerprints, and bounded reads. Dataflow stores only opaque source "
                 "references and expected fingerprints. The first executable provider "
-                "imports immutable JSON or CSV snapshots; database and API providers "
-                "can implement the same capability without changing Dataflow."
+                "imports immutable JSON or CSV snapshots 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"),
@@ -184,6 +199,7 @@ def get_manifest() -> ModuleManifest:
 __all__ = [
     "MODULE_ID",
     "MODULE_VERSION",
+    "DATASOURCE_ORIGIN_INTERFACE_VERSION",
     "TABULAR_SOURCE_INTERFACE_VERSION",
     "get_manifest",
     "manifest",

src/govoplan_connectors/backend/tabular_sources.py

@@ -292,7 +292,16 @@ def _context(
         raise TypeError("Tabular source providers require a SQLAlchemy session.")
     if not isinstance(principal, ApiPrincipal):
         raise TabularSourceAccessError("A tenant API principal is required.")
-    if not (has_scope(principal, required_scope) or has_scope(principal, ADMIN_SCOPE)):
+    accepted_scopes = {required_scope, ADMIN_SCOPE}
+    if required_scope == READ_SCOPE:
+        accepted_scopes.update(
+            {
+                "datasources:catalogue:read",
+                "datasources:source:write",
+                "datasources:source:admin",
+            }
+        )
+    if not any(has_scope(principal, scope) for scope in accepted_scopes):
         raise TabularSourceAccessError(f"Missing scope: {required_scope}")
     return session, principal
 

tests/test_datasource_origins.py

@@ -0,0 +1,87 @@
+from __future__ import annotations
+
+import unittest
+
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker
+
+from govoplan_core.auth import ApiPrincipal
+from govoplan_core.core.access import PrincipalRef
+from govoplan_core.core.datasources import DatasourceOriginReadRequest
+from govoplan_core.core.tabular_sources import TabularSnapshotInput
+from govoplan_core.db.base import Base
+from govoplan_connectors.backend.datasource_origins import (
+    ConnectorDatasourceOriginProvider,
+)
+from govoplan_connectors.backend.db.models import ConnectorTabularSource
+from govoplan_connectors.backend.tabular_sources import (
+    WRITE_SCOPE,
+    SqlTabularSourceProvider,
+)
+
+
+def principal(*, scopes: tuple[str, ...]) -> ApiPrincipal:
+    return ApiPrincipal(
+        principal=PrincipalRef(
+            account_id="account-1",
+            membership_id="membership-1",
+            tenant_id="tenant-1",
+            scopes=frozenset(scopes),
+        ),
+        account=object(),
+        user=object(),
+    )
+
+
+class ConnectorDatasourceOriginTests(unittest.TestCase):
+    def setUp(self) -> None:
+        self.engine = create_engine("sqlite:///:memory:")
+        Base.metadata.create_all(
+            self.engine,
+            tables=[ConnectorTabularSource.__table__],
+        )
+        self.Session = sessionmaker(bind=self.engine)
+        self.session = self.Session()
+        provider = SqlTabularSourceProvider()
+        self.source = provider.create_snapshot(
+            self.session,
+            principal(scopes=(WRITE_SCOPE,)),
+            snapshot=TabularSnapshotInput(
+                name="Imported cases",
+                source_name="imported_cases",
+                rows=({"id": 1, "name": "Ada"},),
+            ),
+        )
+        self.session.commit()
+        self.origins = ConnectorDatasourceOriginProvider(provider)
+
+    def tearDown(self) -> None:
+        self.session.close()
+        Base.metadata.drop_all(
+            self.engine,
+            tables=[ConnectorTabularSource.__table__],
+        )
+        self.engine.dispose()
+
+    def test_datasource_reader_can_discover_and_read_connector_origin(self) -> None:
+        datasource_principal = principal(
+            scopes=("datasources:catalogue:read",),
+        )
+
+        origins = self.origins.list_origins(
+            self.session,
+            datasource_principal,
+        )
+        result = self.origins.read_origin(
+            self.session,
+            datasource_principal,
+            request=DatasourceOriginReadRequest(origin_ref=self.source.ref),
+        )
+
+        self.assertEqual((self.source.ref,), tuple(item.ref for item in origins))
+        self.assertEqual(("live", "cached"), origins[0].supported_modes)
+        self.assertEqual(({"id": 1, "name": "Ada"},), result.rows)
+
+
+if __name__ == "__main__":
+    unittest.main()