diff --git a/harness/tests/test_dwh_adapters.py b/harness/tests/test_dwh_adapters.py new file mode 100644 index 00000000..7fdb8395 --- /dev/null +++ b/harness/tests/test_dwh_adapters.py @@ -0,0 +1,42 @@ +import pytest + +from tht.config import DatabaseConfig, RestConfig +from tht.execute import ExecutionError +from tht.ports import DwhAdapter + + +def postgres_factory(): + from tht.adapters.dwh import PostgresDwhAdapter + + return PostgresDwhAdapter( + DatabaseConfig(database="analytics", schema="dw", user="reader", password="secret") + ) + + +def rest_factory(): + from tht.adapters.dwh import ThothRestDwhAdapter + + database = DatabaseConfig( + database="analytics", + schema="dw", + user="unused", + password="unused", + transport="rest", + ) + return ThothRestDwhAdapter( + database, + RestConfig(base_url="https://dwh.example.test", api_key="secret"), + ) + + +@pytest.mark.parametrize("factory", [postgres_factory, rest_factory]) +def test_adapter_rejects_write_sql_without_using_transport(factory): + with pytest.raises(ExecutionError, match="read-only enforcement"): + factory().run_query("delete from fact_sales") + + +@pytest.mark.parametrize("factory", [postgres_factory, rest_factory]) +def test_adapter_satisfies_dwh_protocol(factory): + adapter = factory() + assert isinstance(adapter, DwhAdapter) + assert adapter.capabilities.introspection is True diff --git a/harness/tht/adapters/dwh/__init__.py b/harness/tht/adapters/dwh/__init__.py new file mode 100644 index 00000000..87e3eacc --- /dev/null +++ b/harness/tht/adapters/dwh/__init__.py @@ -0,0 +1,6 @@ +"""Data-warehouse adapter implementations.""" + +from tht.adapters.dwh.postgres import PostgresDwhAdapter +from tht.adapters.dwh.thoth_rest import ThothRestDwhAdapter + +__all__ = ["PostgresDwhAdapter", "ThothRestDwhAdapter"] diff --git a/harness/tht/adapters/dwh/postgres.py b/harness/tht/adapters/dwh/postgres.py new file mode 100644 index 00000000..abd9d62d --- /dev/null +++ b/harness/tht/adapters/dwh/postgres.py @@ -0,0 +1,41 @@ +"""Direct PostgreSQL implementation of the DWH port.""" + +from tht.config import DatabaseConfig +from tht.db import execute +from tht.db.connection import make_engine, ping +from tht.db.introspect import introspect +from tht.execute import ExecResult, PlanSummary +from tht.mschema.models import PhysicalSchema +from tht.ports.dwh import DwhCapabilities, DwhHealth + + +class PostgresDwhAdapter: + capabilities = DwhCapabilities() + + def __init__(self, config: DatabaseConfig): + self._config = config + self._engine = make_engine(config) + + def health(self) -> DwhHealth: + try: + ping(self._engine) + except Exception as exc: + return DwhHealth(ok=False, detail=str(exc)) + return DwhHealth(ok=True) + + def introspect(self) -> PhysicalSchema: + return introspect(self._engine, self._config.database, self._config.db_schema) + + def run_query(self, sql: str, *, limit: int | None = None) -> ExecResult: + return execute.run_query(self._engine, sql, limit=limit) + + def explain(self, sql: str) -> PlanSummary: + return execute.explain(self._engine, sql) + + def sample_column(self, table: str, column: str, *, limit: int) -> list[object]: + return execute.sample_column( + self._engine, self._config.db_schema, table, column, limit=limit + ) + + def distinct_values(self, table: str, column: str) -> list[object]: + return execute.distinct_values(self._engine, self._config.db_schema, table, column) diff --git a/harness/tht/adapters/dwh/thoth_rest.py b/harness/tht/adapters/dwh/thoth_rest.py new file mode 100644 index 00000000..9105b12d --- /dev/null +++ b/harness/tht/adapters/dwh/thoth_rest.py @@ -0,0 +1,51 @@ +"""Thoth/PostgREST implementation of the DWH port.""" + +from tht.config import DatabaseConfig, RestConfig +from tht.db.introspect import introspect_rest +from tht.execute import ExecResult, PlanSummary +from tht.mschema.models import PhysicalSchema +from tht.ports.dwh import DwhCapabilities, DwhHealth +from tht.rest.client import RestClient, RestError +from tht.rest.execute import ( + distinct_values_rest, + explain_rest, + run_controlled_rest, + sample_column_rest, +) + + +class ThothRestDwhAdapter: + capabilities = DwhCapabilities() + + def __init__(self, database: DatabaseConfig, rest: RestConfig): + self._database = database + self._client = RestClient(rest) + + def health(self) -> DwhHealth: + try: + result = self._client.ping() + except RestError as exc: + return DwhHealth(ok=False, detail=str(exc)) + ok = bool(result.get("db_connected", True)) + return DwhHealth(ok=ok, detail=None if ok else str(result)) + + def introspect(self) -> PhysicalSchema: + return introspect_rest( + self._client, self._database.database, self._database.db_schema + ) + + def run_query(self, sql: str, *, limit: int | None = None) -> ExecResult: + return run_controlled_rest(self._client, sql, limit=10 if limit is None else limit) + + def explain(self, sql: str) -> PlanSummary: + return explain_rest(self._client, sql) + + def sample_column(self, table: str, column: str, *, limit: int) -> list[object]: + return sample_column_rest( + self._client, self._database.db_schema, table, column, limit=limit + ) + + def distinct_values(self, table: str, column: str) -> list[object]: + return distinct_values_rest( + self._client, self._database.db_schema, table, column + ) diff --git a/harness/tht/db/execute.py b/harness/tht/db/execute.py new file mode 100644 index 00000000..36f5253e --- /dev/null +++ b/harness/tht/db/execute.py @@ -0,0 +1,42 @@ +"""PostgreSQL execution operations used by the direct DWH adapter.""" + +from sqlalchemy import Engine + +from tht.execute import ExecResult, PlanSummary, explain as _explain, run_controlled + +DEFAULT_LIMIT = 10 +DEFAULT_TIMEOUT_MS = 30_000 + + +def run_query(engine: Engine, sql: str, *, limit: int | None = None) -> ExecResult: + return run_controlled( + engine, + sql, + limit=DEFAULT_LIMIT if limit is None else limit, + timeout_ms=DEFAULT_TIMEOUT_MS, + ) + + +def explain(engine: Engine, sql: str) -> PlanSummary: + return _explain(engine, sql, timeout_ms=DEFAULT_TIMEOUT_MS) + + +def sample_column( + engine: Engine, schema: str, table: str, column: str, *, limit: int +) -> list[object]: + quote = engine.dialect.identifier_preparer.quote + sql = ( + f"SELECT {quote(column)} FROM {quote(schema)}.{quote(table)} " + f"WHERE {quote(column)} IS NOT NULL" + ) + return [row[0] for row in run_query(engine, sql, limit=limit).rows] + + +def distinct_values(engine: Engine, schema: str, table: str, column: str) -> list[object]: + quote = engine.dialect.identifier_preparer.quote + sql = ( + f"SELECT DISTINCT {quote(column)} FROM {quote(schema)}.{quote(table)} " + f"WHERE {quote(column)} IS NOT NULL ORDER BY {quote(column)}" + ) + result = run_controlled(engine, sql, limit=1000, timeout_ms=DEFAULT_TIMEOUT_MS) + return [row[0] for row in result.rows] diff --git a/harness/tht/rest/execute.py b/harness/tht/rest/execute.py index 223f460f..139fc0e5 100644 --- a/harness/tht/rest/execute.py +++ b/harness/tht/rest/execute.py @@ -46,3 +46,14 @@ def explain_rest(client, sql: str) -> PlanSummary: except RestError as e: raise ExecutionError(str(e)) from e return parse_text_plan(lines) + + +def sample_column_rest( + client, schema: str, table: str, column: str, *, limit: int +) -> list[object]: + rows = client.top_values(schema, table, column, limit) + return [row["value"] for row in rows if row.get("value") is not None] + + +def distinct_values_rest(client, schema: str, table: str, column: str) -> list[object]: + return sample_column_rest(client, schema, table, column, limit=1000)