diff --git a/pkg-py/src/commons/_catalog/_core.py b/pkg-py/src/commons/_catalog/_core.py index 0456919e..f0ebe675 100644 --- a/pkg-py/src/commons/_catalog/_core.py +++ b/pkg-py/src/commons/_catalog/_core.py @@ -76,6 +76,11 @@ class TableRegistry: relations: dict[str, Relation] validate: dict[str, TableId] namespace_selected: bool + # The relations the listing reported and exclude then removed, so a + # caller left with nothing, or with a dictionary entry that now matches + # nothing, can say which it was. A name the warehouse never had is not in + # here: exclude is not what made it absent. + dropped: list[Relation] = field(default_factory=list) @dataclass @@ -221,6 +226,9 @@ def table_registry( relations.extend(list_relations(selector)) keep = excluded([item.name for item in relations], exclude) + dropped = [ + item for item, hidden in zip(relations, keep) if hidden and item.discovered + ] relations = [item for item, hidden in zip(relations, keep) if not hidden] validate = [ table_id @@ -252,6 +260,7 @@ def table_registry( relations=labelled, validate={table_id.label: table_id for table_id in validate}, namespace_selected=namespace_selected, + dropped=dropped, ) @@ -475,14 +484,14 @@ def _match_one( relative = [ label for label in labels - if _has_suffix(relations[label], suffix, identifier_case) + if has_suffix(relations[label], suffix, identifier_case) ] if len(relative) > 1: _abort_ambiguous(authored_name, relative) return relative[0] if relative else None -def _has_suffix( +def has_suffix( relation: Relation, suffix: list[str], identifier_case: str | None ) -> bool: path = relation.id.parts diff --git a/pkg-py/src/commons/_catalog/_import.py b/pkg-py/src/commons/_catalog/_import.py new file mode 100644 index 00000000..696f6413 --- /dev/null +++ b/pkg-py/src/commons/_catalog/_import.py @@ -0,0 +1,228 @@ +"""Turning a warehouse connection into the tables a data source exposes. + +The order is what matters here. The session identity is read first, because +every access answer that follows was decided for it; explicitly named +relations are checked before anything is described, so a name the caller got +wrong fails at construction rather than mid-conversation; and the session is +read again at the end, because a role that moved during discovery invalidates +what was just learned. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Any + +from .._data_source import TableId +from . import _databricks, _snowflake +from ._core import ( + Manifest, + MergedDictionary, + Relation, + Selector, + check_exclude, + has_suffix, + id_type, + merge_dictionary, + table_registry, +) +from ._security import ( + SessionSnapshot, + check_session, + require_queryable, + require_queryable_relations, + session_snapshot, +) + +__all__ = ["ImportedCatalog", "import_catalog", "is_warehouse"] + +# Each warehouse's reader, and the case its identifiers fold to. +_READERS = { + "snowflake": (_snowflake, "upper"), + "databricks": (_databricks, "lower"), +} + + +@dataclass +class ImportedCatalog: + """What a warehouse listing contributes to a data source.""" + + tables: list[str] + table_ids: dict[str, TableId] + relations: dict[str, Relation] + manifest: Manifest + dictionary: Any | None + definition_bindings: dict[str, Any] | None + session: SessionSnapshot | None + + +def is_warehouse(backend: Any) -> bool: + return backend.dialect() in _READERS + + +def import_catalog( + backend: Any, + tables: Any = None, + exclude: list[str] | None = None, + dictionary: Any = None, +) -> ImportedCatalog: + """Resolve a selection against a warehouse and fold it into a dictionary.""" + reader, identifier_case = _READERS[backend.dialect()] + check_exclude(exclude) + session = session_snapshot(backend) + + registry = table_registry( + _selectors(backend, reader, tables), + exact_relation=lambda selector: reader.exact_relation(backend, selector), + list_relations=lambda selector: reader.list_relations(backend, selector), + exclude=exclude, + ) + if registry.validate: + require_queryable_relations(backend, registry.validate, registry.relations) + if not registry.relations: + if registry.dropped: + raise ValueError( + "The resolved catalog selection contains no objects: exclude " + f"dropped every relation it resolved to. Narrow {exclude!r}." + ) + raise ValueError("The resolved catalog selection contains no objects.") + + # A relation named in the selection has just been probed, so the merge + # only has to check the ones the listing brought in. + queryable = set(registry.validate) + + def access_check(table_id: TableId, label: str) -> None: + if label in queryable: + return + require_queryable(backend, table_id, label) + queryable.add(label) + + merged = merge_dictionary( + dictionary, + registry.relations, + describe_relation=lambda table_id: reader.describe_relation(backend, table_id), + identifier_case=identifier_case, + access_check=access_check, + ) + check_session(backend, session) + _check_definitions_bound(merged, registry.dropped, exclude, identifier_case) + + # The manifest starts with every relation unknown, including the ones + # just probed. Carrying the construction-time answer forward would save a + # round trip on first touch at the cost of serving a grant revoked in + # between, and the sibling implementation re-probes for the same reason. + manifest = Manifest.build( + merged.relations, namespace_selected=registry.namespace_selected + ) + return ImportedCatalog( + tables=list(merged.relations), + table_ids={label: item.id for label, item in merged.relations.items()}, + relations=merged.relations, + manifest=manifest, + dictionary=merged.dictionary, + definition_bindings=merged.definition_bindings, + session=session, + ) + + +def _check_definitions_bound( + merged: MergedDictionary, + dropped: list[Relation] | None = None, + exclude: list[str] | None = None, + identifier_case: str | None = None, +) -> None: + """Refuse definitions the merge renamed out from under. + + The merge re-keys an authored dictionary to the warehouse's own labels + and column spellings, but a definition's expression still names what the + author wrote. Lowering it as written would emit SQL against identifiers + the warehouse does not have, so it is refused until the compiler can bind + the two together. + """ + bindings = merged.definition_bindings + exports = getattr(merged.dictionary, "definition_exports", None) or {} + if not bindings or not exports: + return + for authored_table, definitions in exports.items(): + if not definitions: + continue + # An authored table that matched nothing is dropped by the merge, so + # its definitions would go with it and the agent would never be told. + if bindings["tables"].get(authored_table) is None: + if _was_excluded(authored_table, dropped, identifier_case): + raise ValueError( + f"Authored table {authored_table!r} declares definitions, " + f"and exclude dropped it from the catalog listing. Narrow " + f"{exclude!r}, or drop the table from the data dictionary." + ) + raise ValueError( + f"Authored table {authored_table!r} declares definitions, and " + f"does not match an exposed relation. Name it as the data " + f"source selects it, or drop it from the data dictionary." + ) + # Every authored column is checked, not only the ones a definition + # reads: which columns an expression touches is in the compiler's + # parse tree, and the refusal is temporary either way. A column the + # warehouse never reported is caught here too, since a definition + # over it would lower to SQL naming nothing at all. + columns = bindings["columns"].get(authored_table) or {} + unbound = [name for name, discovered in columns.items() if discovered != name] + if unbound: + raise NotImplementedError( + f"Table {authored_table!r} declares definitions, and the " + f"warehouse does not have column {unbound[0]!r} under that " + f"name. Binding a definition to the discovered spelling is " + f"not available yet." + ) + + +def _was_excluded( + authored_table: str, dropped: list[Relation] | None, identifier_case: str | None +) -> bool: + """Whether exclude is what removed the relation an authored name meant. + + Compared against the relations exclude actually dropped rather than + against the patterns: a pattern is written in the warehouse's spelling + and an authored name need not be, so only the folded names line up. The + authored name may be qualified, so it is matched as a suffix, by the + same rule the merge uses to find the relation in the first place. + """ + suffix = authored_table.split(".") + return any( + has_suffix(item, suffix, identifier_case) for item in dropped or [] + ) + + +def _selectors(backend: Any, reader: Any, tables: Any) -> list[Selector]: + """Read a `tables` selection, defaulting to the connection's namespace. + + With nothing named, the schema the connection already points at is the + selection: it is the one namespace the caller has already chosen. + """ + if tables is None: + return [reader.current_namespace(backend)] + entries = list(tables) if isinstance(tables, (list, tuple)) else [tables] + if not entries: + raise ValueError("tables must name at least one relation or namespace.") + return [_selector(entry) for entry in entries] + + +def _selector(entry: Any) -> Selector: + """One selection entry as a `Selector`, whatever it was spelled as. + + A string or a `TableId` is read the way it is everywhere else, as a + relation whose dots qualify it, so the spelling rules and their wording + come from the one place that owns them. A namespace has to be a + `Selector`, because `ANALYTICS.PUBLIC` on its own does not say whether + PUBLIC is a schema or a table. + """ + if isinstance(entry, Selector): + # Raises unless the selector names a relation or a namespace. + id_type(entry) + return entry + from .._data_source import _table_entry_id + + table_id = _table_entry_id(entry) + return Selector( + catalog=table_id.catalog, schema=table_id.schema, table=table_id.table + ) diff --git a/pkg-py/src/commons/_data_source.py b/pkg-py/src/commons/_data_source.py index 5d0583c2..4fe6cc40 100644 --- a/pkg-py/src/commons/_data_source.py +++ b/pkg-py/src/commons/_data_source.py @@ -18,6 +18,7 @@ from ._sql_guard import check_query if TYPE_CHECKING: + from ._catalog import Manifest, Relation, SessionSnapshot from ._data_dictionary import DataDictionary __all__ = ["DataSource", "TableId", "data_source", "list_tables"] @@ -39,6 +40,25 @@ def _fold(name: str) -> str: return name.translate(_ASCII_FOLD) +def _with_compiled_definitions(source: DataSource) -> DataSource: + """Lower a dictionary's governed definitions, once construction is done. + + The dialect and the final table set are only known here, which is why + lowering waits for them. It runs at the end of every constructor rather + than in `data_source()`, so that a source built through a constructor + directly carries its definitions instead of silently dropping them. A + warehouse source folds its catalog listing into the authored dictionary, + so the one to compile against is the source's, not the one passed in. + """ + if source.dictionary is not None: + from ._definitions import attach_compiled_definitions + + attach_compiled_definitions( + source.dictionary, source.dialect(), set(source.tables) + ) + return source + + @dataclass(frozen=True) class TableId: """A table's identity, qualified as far as the backend has levels. @@ -95,6 +115,14 @@ class DataSource: table_ids: dict[str, TableId] = field(default_factory=dict) pending: _PendingPins | None = None dictionary: DataDictionary | None = None + # Set only for a warehouse source: what its catalog reported, what is + # known about access to each relation, the connection identity all of + # that was decided for, and how the merge re-keyed authored names onto + # the warehouse's own. + relations: dict[str, Relation] | None = None + manifest: Manifest | None = None + session: SessionSnapshot | None = None + definition_bindings: dict[str, Any] | None = None @classmethod def from_frames(cls, **frames: Any) -> DataSource: @@ -122,29 +150,101 @@ def from_frames(cls, **frames: Any) -> DataSource: ) @classmethod - def from_engine(cls, engine: sqlalchemy.Engine, tables: Any = None) -> DataSource: + def from_engine( + cls, + engine: sqlalchemy.Engine, + tables: Any = None, + exclude: list[str] | None = None, + dictionary: DataDictionary | None = None, + ) -> DataSource: """Query a caller's database directly. Nothing is copied. - With `tables` unset the backend's own listing is taken as given: it - reports what exists, so there is nothing to check and no round trip - worth paying for. + A Snowflake or Databricks engine imports its catalog: the selection is + resolved against the warehouse, access to it is verified for the + current principal, and what it reports is folded into `dictionary`. + + With `tables` unset on any other backend, its own listing is taken as + given: it reports what exists, so there is nothing to check and no + round trip worth paying for. + + `exclude` takes unqualified object-name globs to drop from a + warehouse catalog listing, such as `"TMP_*"`. Only a warehouse has a + listing to drop from, so any other engine refuses it. + + On a warehouse `dictionary` is taken here because the catalog listing + is folded into it during construction; on any other engine it is + simply attached. Its governed definitions are lowered once the + dialect and the final table set are known, at the end of + construction. """ + from ._catalog._import import is_warehouse + backend = EngineBackend(engine) + if is_warehouse(backend): + return _with_compiled_definitions( + cls._from_warehouse(backend, tables, exclude, dictionary) + ) + if exclude is not None: + raise ValueError( + "exclude selects out of a warehouse catalog listing, and is " + f"supported only for Snowflake and Databricks. This engine is " + f"{backend.dialect()}." + ) if tables is None: discovered = backend.list_tables() - return cls( - backend=backend, - tables=discovered, - table_ids={name: TableId(table=name) for name in discovered}, + return _with_compiled_definitions( + cls( + backend=backend, + tables=discovered, + table_ids={name: TableId(table=name) for name in discovered}, + dictionary=dictionary, + ) ) registry = normalize_table_registry(tables) _check_tables_exist(backend, registry) - return cls(backend=backend, tables=list(registry), table_ids=registry) + return _with_compiled_definitions( + cls( + backend=backend, + tables=list(registry), + table_ids=registry, + dictionary=dictionary, + ) + ) @classmethod - def from_board(cls, board: Any, tables: Any) -> DataSource: - """Expose a pins board's pins as tables, each read on first use.""" + def _from_warehouse( + cls, + backend: Backend, + tables: Any, + exclude: list[str] | None, + dictionary: DataDictionary | None, + ) -> DataSource: + from ._catalog._import import import_catalog + + imported = import_catalog(backend, tables, exclude, dictionary) + return cls( + backend=backend, + tables=imported.tables, + table_ids=imported.table_ids, + dictionary=imported.dictionary, + relations=imported.relations, + manifest=imported.manifest, + session=imported.session, + definition_bindings=imported.definition_bindings, + ) + + @classmethod + def from_board( + cls, board: Any, tables: Any, dictionary: DataDictionary | None = None + ) -> DataSource: + """Expose a pins board's pins as tables, each read on first use. + + `dictionary` is taken here so that the argument survives the + dispatcher; a board has no catalog listing to fold into it. Its + governed definitions are lowered at the end of construction, once + the dialect and the final table set are known. + """ if not isinstance(tables, dict): raise TypeError( "For a pins board, tables must be a mapping of table name to " @@ -171,15 +271,27 @@ def from_board(cls, board: Any, tables: Any) -> DataSource: labels = list(tables) _check_labels_distinct(labels) _check_labels_free(con, labels) - return cls( - backend=DuckDBBackend(con), - tables=labels, - table_ids={label: TableId(table=label) for label in labels}, - pending=_PendingPins(board=board, pins=dict(tables)), + return _with_compiled_definitions( + cls( + backend=DuckDBBackend(con), + tables=labels, + table_ids={label: TableId(table=label) for label in labels}, + pending=_PendingPins(board=board, pins=dict(tables)), + dictionary=dictionary, + ) ) def query(self, sql: str) -> list[dict[str, Any]]: - """Run one read-only statement, rejecting anything else first.""" + """Run one read-only statement, rejecting anything else first. + + On a warehouse source the connection identity is checked before the + statement is read: access to these tables was decided for one + principal, role, and namespace, so a query raises rather than runs + once any of those has moved. + """ + from ._catalog import check_session + + check_session(self.backend, self.session) check_query(sql, dialect=self.backend.dialect()) if self.pending is None: return self.backend.query(sql) @@ -255,6 +367,7 @@ def dialect(self) -> str: def data_source( *args: Any, tables: Any = None, + exclude: Any = None, dictionary: Any = None, **frames: Any, ) -> DataSource: @@ -262,17 +375,22 @@ def data_source( A thin dispatcher over the constructors, which are the documented way in. - `tables` and `dictionary` are keyword-only options, so both names are - reserved in every form: a frame passed under either name is rejected - with a TypeError naming it, never silently consumed. `tables` selects - tables of the engine and board forms; `dictionary` attaches a data - dictionary to any form. To use either as a frame name, call + `tables`, `exclude`, and `dictionary` are keyword-only options, so all + three names are reserved in every form: a frame passed under one of them + is rejected with a TypeError naming it, never silently consumed. + `tables` selects tables of the engine and board forms, `exclude` drops + relations from a warehouse catalog listing by glob, and `dictionary` + attaches a data dictionary to any form. To use one as a frame name, call `DataSource.from_frames()` directly. A dictionary's governed definitions are compiled for the source's - dialect here, so construction raises if the dialect has no emitter - (only DuckDB does today), if a definition sits on a table the source - does not expose, or if a metric mixes row and aggregate grain. + dialect during construction, so construction raises if the dialect has + no emitter (DuckDB, Snowflake, and Databricks have one), if a definition + sits on a table the source does not expose, or if a metric mixes row and + aggregate grain. On a warehouse it also raises if a table declaring + definitions matched no exposed relation, and, until the compiler can + bind an authored name to the discovered one, if the warehouse spells one + of that table's columns differently. """ from ._data_dictionary import as_data_dictionary @@ -288,29 +406,41 @@ def data_source( "Pass either a connection or named data frames, not both. " f"Got a positional argument and the frames {sorted(frames)}." ) - source = _from_positional(args[0], tables) + source = _from_positional(args[0], tables, exclude, resolved) else: if tables is not None: raise TypeError( "`tables` selects tables of an engine or pins board; with " "named frames there is nothing for it to select." ) + if exclude is not None: + raise TypeError( + "`exclude` drops relations from a warehouse catalog listing; " + "with named frames there is no listing to drop them from." + ) source = DataSource.from_frames(**frames) + source.dictionary = resolved - source.dictionary = resolved - if resolved is not None: - # The dialect is only known now, which is why lowering waits for it. - from ._definitions import attach_compiled_definitions - - attach_compiled_definitions(resolved, source.dialect(), set(source.tables)) + # The frames form attaches its dictionary here rather than in the + # constructor, so this is where its definitions can be lowered. + source = _with_compiled_definitions(source) return source -def _from_positional(value: Any, tables: Any) -> DataSource: +def _from_positional( + value: Any, tables: Any, exclude: Any = None, dictionary: Any = None +) -> DataSource: if isinstance(value, sqlalchemy.Engine): - return DataSource.from_engine(value, tables=tables) + return DataSource.from_engine( + value, tables=tables, exclude=exclude, dictionary=dictionary + ) if hasattr(value, "pin_list") and hasattr(value, "pin_read"): - return DataSource.from_board(value, tables) + if exclude is not None: + raise TypeError( + "`exclude` drops relations from a warehouse catalog listing; " + "a pins board has no listing to drop them from." + ) + return DataSource.from_board(value, tables, dictionary=dictionary) raise TypeError( "data_source() takes a SQLAlchemy Engine, a pins board, or named data " f"frames. Got {type(value).__name__}." diff --git a/pkg-py/src/commons/_definitions/_compile.py b/pkg-py/src/commons/_definitions/_compile.py index cb2551c3..655683ce 100644 --- a/pkg-py/src/commons/_definitions/_compile.py +++ b/pkg-py/src/commons/_definitions/_compile.py @@ -95,8 +95,12 @@ def attach_compiled_definitions( getattr(dictionary, "definition_exports", None) or {} ) compiled: dict[str, list[ExportRecord]] = {} - for table_name in dictionary.tables: - definitions = exports.get(table_name) or {} + for table_name, entry in dictionary.tables.items(): + # A catalog import re-keys the dictionary to the warehouse's labels + # and records what the author called each table, which is how its + # exports are still found afterwards. + authored = getattr(entry, "authored_name", None) or table_name + definitions = exports.get(authored) or {} if not definitions: compiled[table_name] = [] continue diff --git a/pkg-py/tests/_warehouse.py b/pkg-py/tests/_warehouse.py new file mode 100644 index 00000000..92e689eb --- /dev/null +++ b/pkg-py/tests/_warehouse.py @@ -0,0 +1,178 @@ +"""A warehouse backend that answers from canned rows. + +It stands in for the network, not for a warehouse's semantics: the row shapes +come from `tests/shared/catalog-rows.json`, which is the file that governs how +both packages read a listing, and refusing a relation is how a real warehouse +refuses one, so what is exercised is commons' handling of both. + +`FakeWarehouse` speaks Snowflake and `FakeDatabricks` speaks Databricks. They +answer different queries because the two readers ask different ones, and a +fake that answered both would hide a reader wired to the wrong warehouse. +""" + +from commons._data_source import TableId +from tests._shared import load_shared_fixture + +__all__ = [ + "FakeDatabricks", + "FakeWarehouse", + "column_row", + "databricks_column_row", + "databricks_relation_row", + "relation_row", +] + + +def _shape(backend: str, group: str, accept) -> dict: + """A row of the shared fixture the readers accept, as a template to vary. + + Taking the shape from the fixture rather than restating it means a change + to what a warehouse returns reaches these fakes instead of leaving them + answering with a shape the readers no longer expect. + """ + rows = load_shared_fixture("catalog-rows")[backend][group]["rows"] + for row in rows: + if accept(row): + return dict(row) + raise AssertionError(f"no usable {backend} {group} row") + + +def relation_row(name: str, catalog: str = "ANALYTICS", schema: str = "PUBLIC") -> dict: + row = _shape("snowflake", "relations", lambda row: row["kind"] == "TABLE") + row.update(name=name, database_name=catalog, schema_name=schema, comment="") + return row + + +def column_row(name: str) -> dict: + row = _shape("snowflake", "columns", lambda row: row["kind"] == "COLUMN") + row.update(name=name, type="NUMBER(38,0)", comment="") + row["null?"] = "N" + return row + + +def databricks_relation_row(name: str, catalog: str = "main", schema: str = "sales"): + row = _shape("databricks", "relations", lambda row: row["table_type"] != "VIEW") + row.update( + table_catalog=catalog, table_schema=schema, table_name=name, comment="" + ) + return row + + +def databricks_column_row(name: str) -> dict: + # A DESCRIBE reply runs on past the columns, so the template is a row + # from before the metadata marker. + row = _shape( + "databricks", + "columns", + lambda row: row["col_name"] not in ("", "# Partition Information"), + ) + row.update(col_name=name, data_type="bigint", comment="") + return row + + +class _Recorder: + def __init__(self): + self.queries: list[str] = [] + self.refuse: set[str] = set() + + def _probe(self, sql: str): + for name in self.refuse: + if name in sql: + raise PermissionError(f"Insufficient privileges on {name}") + return [] + + def quote(self, table_id: TableId) -> str: + return ".".join(f'"{part}"' for part in table_id.parts) + + def inspector(self): + return None + + +class FakeWarehouse(_Recorder): + """A Snowflake standing in for the network, answering from canned rows.""" + + def __init__( + self, dialect="snowflake", relations=None, role="REPORTER", columns=("ID",) + ): + super().__init__() + self._dialect = dialect + self._relations = relations if relations is not None else ["SALES", "ORDERS"] + self._columns = list(columns) + self.role = role + + def query(self, sql: str): + self.queries.append(sql) + if sql.startswith("SELECT CURRENT_USER()"): + return [ + { + "PRINCIPAL": "ANALYST", + "ROLE": self.role, + "SECONDARY_ROLES": "ALL", + "CATALOG": "ANALYTICS", + "SCHEMA": "PUBLIC", + } + ] + if sql.startswith("SELECT CURRENT_DATABASE()"): + return [{"CATALOG": "ANALYTICS", "SCHEMA": "PUBLIC"}] + if sql.startswith("SHOW OBJECTS"): + return [ + relation_row(name) + for name in self._relations + if f"'{name}'" in sql or "LIKE" not in sql + ] + if sql.startswith("DESC TABLE"): + return [column_row(name) for name in self._columns] + if sql.startswith("SELECT * FROM"): + return self._probe(sql) + raise AssertionError(f"unexpected query: {sql}") + + def list_tables(self) -> list[str]: + return list(self._relations) + + def dialect(self) -> str: + return self._dialect + + +class FakeDatabricks(_Recorder): + """A Databricks standing in for the network. + + Unity Catalog only: `hive_metastore` takes a different listing path and + nothing here selects it. + """ + + def __init__(self, relations=None, columns=("id",)): + super().__init__() + self._relations = relations if relations is not None else ["sales", "orders"] + self._columns = list(columns) + + def query(self, sql: str): + self.queries.append(sql) + if sql.startswith("SELECT CURRENT_USER()"): + return [ + {"principal": "analyst@example.com", "catalog": "main", "schema": "sales"} + ] + if sql.startswith("SELECT CURRENT_CATALOG()"): + return [{"catalog": "main", "schema": "sales"}] + if "system.information_schema.columns" in sql: + # DESCRIBE TABLE does not report nullability, so the reader asks + # the information schema for it separately. + return [ + {"column_name": name, "is_nullable": "YES"} for name in self._columns + ] + if "system.information_schema.tables" in sql: + return [ + databricks_relation_row(name) + for name in self._relations + if f"'{name}'" in sql or "table_name =" not in sql + ] + if sql.startswith("DESCRIBE TABLE"): + return [databricks_column_row(name) for name in self._columns] + if sql.startswith("SELECT * FROM"): + return self._probe(sql) + raise AssertionError(f"unexpected query: {sql}") + + def list_tables(self) -> list[str]: + return list(self._relations) + + def dialect(self) -> str: + return "databricks" diff --git a/pkg-py/tests/test_catalog_import.py b/pkg-py/tests/test_catalog_import.py new file mode 100644 index 00000000..f44c1e95 --- /dev/null +++ b/pkg-py/tests/test_catalog_import.py @@ -0,0 +1,303 @@ +"""Importing a warehouse catalog for a data source. + +The reader queries are already covered per backend; what is checked here is +the order the import puts them in, since that order is what makes the source +safe: nothing is described before its access is verified, and the session is +the same one at the end as at the start. +""" + +import pytest + +from commons._catalog import CatalogSessionChangedError, Selector +from commons._catalog._import import import_catalog, is_warehouse +from commons._data_dictionary import DataDictionary +from commons._data_source import TableId +from tests._warehouse import FakeDatabricks, FakeWarehouse + + +def test_only_warehouse_backends_import_a_catalog(): + assert is_warehouse(FakeWarehouse()) + assert is_warehouse(FakeDatabricks()) + assert not is_warehouse(FakeWarehouse(dialect="duckdb")) + + +def test_an_unset_selection_takes_the_current_namespace(): + backend = FakeWarehouse() + + imported = import_catalog(backend) + + assert imported.tables == ["ANALYTICS.PUBLIC.SALES", "ANALYTICS.PUBLIC.ORDERS"] + assert imported.session is not None + assert imported.session.principal == "ANALYST" + + +def test_a_namespace_is_named_with_a_selector(): + backend = FakeWarehouse() + + imported = import_catalog(backend, Selector(catalog="ANALYTICS", schema="PUBLIC")) + + assert list(imported.relations) == [ + "ANALYTICS.PUBLIC.SALES", + "ANALYTICS.PUBLIC.ORDERS", + ] + assert imported.manifest.access == dict.fromkeys(imported.tables, "unknown") + + +def test_a_dotted_string_names_one_relation(): + backend = FakeWarehouse() + + imported = import_catalog(backend, "ANALYTICS.PUBLIC.SALES") + + assert imported.tables == ["ANALYTICS.PUBLIC.SALES"] + + +def test_a_named_relation_is_access_checked_before_the_source_exists(): + backend = FakeWarehouse() + backend.refuse = {"SALES"} + + with pytest.raises(Exception, match="not authorized"): + import_catalog(backend, "ANALYTICS.PUBLIC.SALES") + + +def test_construction_probes_are_not_carried_into_the_manifest(): + # Every relation starts unknown, including the ones construction just + # probed. Carrying that answer forward would serve a grant revoked + # between construction and first touch, so first touch re-probes. + imported = import_catalog(FakeWarehouse(), "ANALYTICS.PUBLIC.SALES") + + assert imported.manifest.access == {"ANALYTICS.PUBLIC.SALES": "unknown"} + + +def test_a_selection_resolving_to_nothing_is_an_error(): + backend = FakeWarehouse(relations=[]) + + with pytest.raises(ValueError, match="no objects"): + import_catalog(backend) + + +def test_an_excluded_relation_is_left_out(): + backend = FakeWarehouse() + + imported = import_catalog(backend, exclude=["ORD*"]) + + assert imported.tables == ["ANALYTICS.PUBLIC.SALES"] + + +def test_a_session_that_moves_during_discovery_is_refused(): + class Moving(FakeWarehouse): + def query(self, sql): + rows = super().query(sql) + if sql.startswith("SELECT CURRENT_USER()"): + self.role = "ADMIN" + return rows + + with pytest.raises(CatalogSessionChangedError): + import_catalog(Moving()) + + +def test_an_authored_dictionary_is_rekeyed_to_the_warehouse_labels(): + backend = FakeWarehouse() + dictionary = DataDictionary.model_validate( + {"tables": [{"name": "sales", "description": "Authored prose"}]} + ) + + imported = import_catalog(backend, dictionary=dictionary) + + assert imported.dictionary is not None + assert list(imported.dictionary.tables) == ["ANALYTICS.PUBLIC.SALES"] + table = imported.dictionary.tables["ANALYTICS.PUBLIC.SALES"] + assert table.description == "Authored prose" + assert list(table.columns) == ["ID"] + + +def test_an_unauthorized_table_stops_the_dictionary_merge(): + backend = FakeWarehouse() + backend.refuse = {"SALES"} + dictionary = DataDictionary.model_validate( + {"tables": [{"name": "sales", "description": "Prose"}]} + ) + + with pytest.raises(Exception, match="not authorized"): + import_catalog(backend, dictionary=dictionary) + + +def _prose() -> DataDictionary: + """A dictionary that matches SALES, so the merge describes it.""" + return DataDictionary.model_validate( + {"tables": [{"name": "sales", "description": "Authored prose"}]} + ) + + +def _kinds(queries: list[str]) -> list[str]: + """Each query as the step it belongs to, so order can be asserted.""" + steps = [] + for sql in queries: + if sql.startswith(("SELECT CURRENT_USER()",)): + steps.append("session") + elif sql.startswith(("SELECT CURRENT_DATABASE()", "SELECT CURRENT_CATALOG()")): + steps.append("namespace") + elif sql.startswith("SHOW OBJECTS") or "information_schema.tables" in sql: + steps.append("list") + elif ( + sql.startswith(("DESC TABLE", "DESCRIBE TABLE")) + or "information_schema.columns" in sql + ): + steps.append("describe") + elif sql.startswith("SELECT * FROM"): + steps.append("probe") + else: + raise AssertionError(f"unclassified query: {sql}") + return steps + + +def test_the_session_is_read_before_anything_it_decides(): + backend = FakeWarehouse() + + import_catalog(backend, "ANALYTICS.PUBLIC.SALES") + + # Every access answer that follows was decided for this identity, so + # reading it second would be reading it about a different connection. + assert _kinds(backend.queries)[0] == "session" + + +def test_a_named_relation_is_probed_before_it_is_described(): + backend = FakeWarehouse() + + import_catalog(backend, "ANALYTICS.PUBLIC.SALES", dictionary=_prose()) + + steps = _kinds(backend.queries) + # A name the caller got wrong should fail at construction, which only + # holds if the probe precedes the describe that would otherwise reveal it. + assert steps.index("probe") < steps.index("describe") + + +def test_the_session_is_read_again_after_discovery(): + backend = FakeWarehouse() + + import_catalog(backend, "ANALYTICS.PUBLIC.SALES", dictionary=_prose()) + + steps = _kinds(backend.queries) + # A role that moved during discovery invalidates what was just learned, + # so the second read has to come after the last thing it invalidates. + assert steps[-1] == "session" + assert steps.count("session") == 2 + assert steps.index("describe") < len(steps) - 1 + + +def test_a_databricks_selection_is_imported_the_same_way(): + backend = FakeDatabricks() + + imported = import_catalog(backend, dictionary=_databricks_prose()) + + assert imported.tables == ["main.sales.sales", "main.sales.orders"] + assert imported.session is not None + assert imported.session.principal == "analyst@example.com" + # Databricks reports no role, so there is none to snapshot. + assert imported.session.role is None + assert _kinds(backend.queries)[0] == "session" + assert _kinds(backend.queries)[-1] == "session" + + +def test_a_databricks_authored_name_matches_whatever_its_case(): + # Databricks reports lower-cased identifiers, so an authored name in any + # other case still has to find the relation it describes. + backend = FakeDatabricks() + + imported = import_catalog(backend, dictionary=_databricks_prose("SALES")) + + assert imported.dictionary is not None + table = imported.dictionary.tables["main.sales.sales"] + assert table.description == "Authored prose" + + +def test_a_databricks_relation_is_probed_before_it_is_described(): + backend = FakeDatabricks() + + import_catalog(backend, "main.sales.orders", dictionary=_databricks_prose("orders")) + + steps = _kinds(backend.queries) + assert steps.index("probe") < steps.index("describe") + + +def _databricks_prose(name: str = "sales") -> DataDictionary: + return DataDictionary.model_validate( + {"tables": [{"name": name, "description": "Authored prose"}]} + ) + + +def test_an_empty_selection_is_refused(): + with pytest.raises(ValueError, match="at least one relation or namespace"): + import_catalog(FakeWarehouse(), []) + + +def test_a_selection_entry_with_an_empty_component_is_refused(): + with pytest.raises(ValueError, match="empty name components"): + import_catalog(FakeWarehouse(), "ANALYTICS..SALES") + + +def test_a_selection_entry_of_the_wrong_type_is_refused(): + with pytest.raises(TypeError, match="table name or a TableId"): + import_catalog(FakeWarehouse(), 5) + + +def test_a_table_id_names_one_relation(): + imported = import_catalog( + FakeWarehouse(), TableId(catalog="ANALYTICS", schema="PUBLIC", table="SALES") + ) + + assert imported.tables == ["ANALYTICS.PUBLIC.SALES"] + + +def test_a_named_relation_the_warehouse_lacks_fails_construction(): + with pytest.raises(ValueError, match="does not have"): + import_catalog(FakeWarehouse(), "ANALYTICS.PUBLIC.MISSING") + + +def test_exclude_that_empties_the_selection_says_so(): + with pytest.raises(ValueError, match="dropped every relation"): + import_catalog(FakeWarehouse(), exclude=["*"]) + + +def test_an_exclude_that_matched_nothing_does_not_blame_exclude(): + # The namespace was already empty, so exclude is not what emptied it. + backend = FakeWarehouse(relations=[]) + + with pytest.raises(ValueError, match="contains no objects") as refusal: + import_catalog(backend, exclude=["NOTHING_*"]) + + assert "exclude dropped" not in str(refusal.value) + + +@pytest.mark.parametrize( + "authored", ["sales", "PUBLIC.sales", "ANALYTICS.PUBLIC.sales"] +) +def test_definitions_on_an_excluded_table_name_exclude_however_qualified(authored): + # An authored name may be qualified, and a glob may not be, so the two + # are matched by the same suffix rule the merge uses. + dictionary = DataDictionary.model_validate( + { + "tables": [ + { + "name": authored, + "columns": [{"name": "id", "type": "number(quantity)"}], + "definitions": [{"name": "total", "expr": "sum(id)"}], + } + ] + } + ) + + with pytest.raises(ValueError, match="exclude dropped it"): + import_catalog(FakeWarehouse(), exclude=["SALES"], dictionary=dictionary) + + + + +def test_excluding_a_name_the_warehouse_never_had_does_not_blame_exclude(): + # The relation was absent, not dropped, so exclude is not the reason + # there is nothing left to expose. + with pytest.raises(ValueError, match="contains no objects") as refusal: + import_catalog( + FakeWarehouse(), "ANALYTICS.PUBLIC.MISSING", exclude=["MISSING"] + ) + + assert "exclude dropped" not in str(refusal.value) diff --git a/pkg-py/tests/test_data_source_warehouse.py b/pkg-py/tests/test_data_source_warehouse.py new file mode 100644 index 00000000..708bbf9e --- /dev/null +++ b/pkg-py/tests/test_data_source_warehouse.py @@ -0,0 +1,161 @@ +"""A data source over a warehouse connection. + +What the catalog import produces is covered against the import itself; this +is about the source it becomes: which tables it exposes, what it remembers +about access, and what it refuses once the connection is no longer the one +the catalog was read for. +""" + +import pytest +import sqlalchemy + +from commons import data_source +from commons._catalog import CatalogSessionChangedError +from commons._data_dictionary import DataDictionary +from commons._data_source import DataSource +from tests._warehouse import FakeWarehouse + + +def warehouse_source(backend=None, **kwargs) -> DataSource: + return DataSource._from_warehouse( + backend or FakeWarehouse(), + kwargs.pop("tables", None), + kwargs.pop("exclude", None), + kwargs.pop("dictionary", None), + ) + + +def test_a_warehouse_source_exposes_its_catalog_labels(): + source = warehouse_source() + + assert source.tables == ["ANALYTICS.PUBLIC.SALES", "ANALYTICS.PUBLIC.ORDERS"] + assert source.table_ids["ANALYTICS.PUBLIC.SALES"].catalog == "ANALYTICS" + assert source.manifest is not None + assert source.relations is not None + + +def test_a_warehouse_source_carries_the_session_it_was_built_for(): + backend = FakeWarehouse() + source = warehouse_source(backend) + + assert source.session is not None + + backend.role = "ADMIN" + with pytest.raises(CatalogSessionChangedError): + source.query("SELECT * FROM sales") + + +def test_a_warehouse_engine_reaches_the_catalog_import(monkeypatch): + # The backend is the seam: everything above it is the real dispatch, from + # data_source() through from_engine()'s warehouse branch. + backend = FakeWarehouse() + monkeypatch.setattr("commons._data_source.EngineBackend", lambda engine: backend) + + source = data_source(sqlalchemy.create_engine("sqlite://"), exclude=["ORD*"]) + + assert source.tables == ["ANALYTICS.PUBLIC.SALES"] + assert source.session is not None + assert source.manifest is not None + + +def test_a_source_over_anything_else_has_no_session_to_check(tmp_path): + engine = sqlalchemy.create_engine(f"sqlite:///{tmp_path / 'db.sqlite'}") + with engine.begin() as connection: + connection.execute(sqlalchemy.text("CREATE TABLE sales (id INTEGER)")) + source = DataSource.from_engine(engine) + + assert source.session is None + assert source.query("SELECT * FROM sales") == [] + + +def test_exclude_is_refused_by_a_backend_with_no_catalog_listing(tmp_path): + engine = sqlalchemy.create_engine(f"sqlite:///{tmp_path / 'db.sqlite'}") + + with pytest.raises(ValueError, match="Snowflake and Databricks"): + DataSource.from_engine(engine, exclude=["staging_*"]) + + +def test_exclude_is_refused_for_a_pins_board(): + class Board: + def pin_list(self): + return ["sales"] + + def pin_read(self, name): + raise AssertionError("not reached") + + with pytest.raises(TypeError, match="warehouse catalog listing"): + data_source(Board(), tables={"sales": "sales"}, exclude=["tmp_*"]) + + +def test_exclude_is_refused_for_named_frames(): + import pandas as pd + + with pytest.raises(TypeError, match="no listing to drop them from"): + data_source(sales=pd.DataFrame({"id": [1]}), exclude=["tmp_*"]) + + +def test_definitions_the_warehouse_spells_differently_are_refused(): + # Until the compiler can bind an authored name to the discovered one, a + # definition over a renamed column would lower to SQL naming a column + # that is not there, so construction fails instead. + dictionary = DataDictionary.model_validate( + { + "tables": [ + { + "name": "sales", + "columns": [{"name": "id", "type": "number(quantity)"}], + "definitions": [{"name": "total", "expr": "sum(id)"}], + } + ] + } + ) + + with pytest.raises(NotImplementedError, match="does not have column"): + warehouse_source(dictionary=dictionary) + + +def sales_dictionary(**table): + return DataDictionary.model_validate( + { + "tables": [ + { + "name": "sales", + "columns": [{"name": "id", "type": "number(quantity)"}], + **table, + } + ] + } + ) + + +def test_a_definition_survives_the_rekeying_a_merge_does(monkeypatch): + # The merge re-keys the dictionary to the warehouse's label, and the + # definition's export records are still keyed by the authored name. + backend = FakeWarehouse(columns=["id"]) + monkeypatch.setattr("commons._data_source.EngineBackend", lambda engine: backend) + dictionary = sales_dictionary(definitions=[{"name": "total", "expr": "sum(id)"}]) + + source = data_source(sqlalchemy.create_engine("sqlite://"), dictionary=dictionary) + + assert source.dictionary is not None + entry = source.dictionary.tables["ANALYTICS.PUBLIC.SALES"] + assert [record.name for record in entry.compiled_definitions] == ["total"] + + +def test_a_definition_on_a_table_that_matched_nothing_is_refused(monkeypatch): + backend = FakeWarehouse(columns=["id"]) + monkeypatch.setattr("commons._data_source.EngineBackend", lambda engine: backend) + dictionary = DataDictionary.model_validate( + { + "tables": [ + { + "name": "unmatched", + "columns": [{"name": "id", "type": "number(quantity)"}], + "definitions": [{"name": "total", "expr": "sum(id)"}], + } + ] + } + ) + + with pytest.raises(ValueError, match="does not match an exposed relation"): + data_source(sqlalchemy.create_engine("sqlite://"), dictionary=dictionary)