diff --git a/README.md b/README.md index ffd4885..462f0f8 100644 --- a/README.md +++ b/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: diff --git a/docs/CONCEPT.md b/docs/CONCEPT.md index 9a6b190..2fd20e9 100644 --- a/docs/CONCEPT.md +++ b/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 diff --git a/docs/CONNECTOR_SOURCE_LIFECYCLE.md b/docs/CONNECTOR_SOURCE_LIFECYCLE.md index b338b28..b40c99e 100644 --- a/docs/CONNECTOR_SOURCE_LIFECYCLE.md +++ b/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` diff --git a/src/govoplan_connectors/backend/datasource_origins.py b/src/govoplan_connectors/backend/datasource_origins.py new file mode 100644 index 0000000..2b6a970 --- /dev/null +++ b/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"] diff --git a/src/govoplan_connectors/backend/manifest.py b/src/govoplan_connectors/backend/manifest.py index 54dc5b2..591568d 100644 --- a/src/govoplan_connectors/backend/manifest.py +++ b/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", diff --git a/src/govoplan_connectors/backend/tabular_sources.py b/src/govoplan_connectors/backend/tabular_sources.py index a515542..eb72af9 100644 --- a/src/govoplan_connectors/backend/tabular_sources.py +++ b/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 diff --git a/tests/test_datasource_origins.py b/tests/test_datasource_origins.py new file mode 100644 index 0000000..c61be8a --- /dev/null +++ b/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()