diff --git a/.env.example b/.env.example index a789212..1544ee0 100644 --- a/.env.example +++ b/.env.example @@ -1,4 +1,8 @@ HOTDATA_API_KEY= +# Existing managed database to load into, addressed by id. Printed on first-run +# create; set it here to reuse the same database on subsequent runs. +HOTDATA_DATABASE_ID= +# Display label used only when creating a new managed database (not a lookup key). HOTDATA_DATABASE=dlt HOTDATA_SCHEMA=public HOTDATA_WRITE_DISPOSITION=append diff --git a/CHANGELOG.md b/CHANGELOG.md index 07d76ad..68378f2 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,14 +7,21 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +## [0.11.0] - 2026-07-24 + +### Added + +- `database_id` param (`hotdata(database_id=...)`) / `HOTDATA_DATABASE_ID` env / `[destination.hotdata] database_id` config — target an existing managed database by id. On a first run with no id, the database is created by its `database_name` label and the **new id is logged** so it can be pinned (`created managed database … set database_id= …`) to reuse the same database on subsequent runs. + ### Changed +- **Breaking:** managed databases are now addressed strictly **by id**, never by name. Hotdata database names are not unique (the `name` field is a display label / description, not an identifier), so the destination no longer lists databases and matches on the name — a practice that could silently read from, write to, or **drop** the wrong database on a collision. An existing database is bound by id via `GET /databases/{id}`; with no id and `create_database_if_missing`, one is created (labelled `database_name`) and addressed by its returned id for the run. Consequence: to load into the same database across runs you must pin its `database_id` (printed on first-run create); without it, each run creates a new database. Supersedes the interim by-name collision guard. **Requires `hotdata-framework>=0.9.0`** (managed-table ops accept a resolved `ManagedDatabase` and skip the read probe — hotdata-dev/sdk-python-framework#52). +- Create/upload/query-scoped API keys (forbidden from reading `/databases`) can bootstrap a managed database as a direct consequence: with no `database_id` the create path issues no read at all, so the key only makes the create it *is* permitted to make, and the returned record is reused (by id) for every subsequent load/add/query in the run. - **Breaking:** `workspace_id` moved out of `HotdataCredentials` (authentication) to a top-level `hotdata(workspace_id=...)` param / `HotdataClientConfiguration` field (configuration), matching the SDK's `Configuration(api_key=, workspace_id=)` shape. It is a **param with no environment-variable fallback** — the `HOTDATA_WORKSPACE` env var is no longer read on this path (the API key remains env-backed, as a secret). Passing `workspace_id` inside a `credentials={...}` dict still works but is deprecated (hoisted with a `DeprecationWarning`); constructing `HotdataCredentials(workspace_id=...)` now raises `TypeError` — pass `workspace_id=` to `hotdata(...)` instead. ### Fixed - The API key now resolves from `HOTDATA_API_KEY` following the README quickstart: setting the env var (with no `credentials=` argument) populates the destination, instead of leaving `credentials.api_key` unset and failing with an opaque `NoneType` error deep in `hotdata-framework`. Missing `api_key`/`workspace_id` now raises a clear `ConfigurationValueError` at setup naming the missing field. -- Managed-database name resolution is now collision-safe and resolved once per run. Hotdata database names are not unique, and the destination previously took the first `list_databases` match on every operation — so a name collision could silently read from, write to, or **drop** the wrong database. It now raises a clear error when a name matches more than one database, resolves the name to its record a single time per run (cached on the shared config), and addresses every subsequent operation (load/add/list/query/drop) by id. ## [0.10.0] - 2026-07-20 diff --git a/README.md b/README.md index 5bc43c8..359685a 100644 --- a/README.md +++ b/README.md @@ -85,7 +85,14 @@ export HOTDATA_API_KEY=your_api_key The API key is a secret, so it's read from the environment (or a dlt secrets provider). The workspace ID is routing, not a secret — pass it as the `workspace_id=` parameter (switching workspaces is a one-line change). -That's it. On first run, the `sales` managed database is created automatically and the `orders` table is loaded. +That's it. On first run, a managed database labelled `sales` is created automatically, the `orders` table is loaded, and the new database **id** is printed: + +``` +hotdata: created managed database db_abc123 (name='sales'). Pin it for future runs by +setting database_id=db_abc123 (HOTDATA_DATABASE_ID / [destination.hotdata] database_id). +``` + +Managed databases are addressed by id — Hotdata database names are not unique, so a name can't identify one. To keep loading into the **same** database on later runs, pass that id (via `hotdata(database_id="db_abc123")`, `HOTDATA_DATABASE_ID`, or `[destination.hotdata] database_id` in `.dlt/config.toml`). Without a pinned id, each run creates a new database. `hotdata` supports nested/child tables, preserves dlt's internal columns (`_dlt_id`, `_dlt_load_id`), and persists schema-version, load, and pipeline-state tables in the managed database so **incremental sources resume correctly across runs**. If an existing managed database is missing a declared table on a later run, the table is added to it in place; existing tables and their data are left untouched. @@ -189,7 +196,8 @@ Where `hotdata` stands against the [dlt destination capability spec](https://dlt |-----------|-------------|---------|-------------| | `api_key` | `HOTDATA_API_KEY` | required | Your Hotdata API key (a secret; passed via `credentials=` or the env var) | | `workspace_id` | — | required | Your Hotdata workspace ID — pass as the `hotdata(workspace_id=...)` param (no env var) | -| `database_name` | `HOTDATA_DATABASE` | `dlt` | Managed database to load into | +| `database_id` | `HOTDATA_DATABASE_ID` | — | Id of an existing managed database to load into. This is how a database is targeted — names are not unique, so the **id** is the identifier. Printed on first-run create; pin it to reuse the database on later runs | +| `database_name` | `HOTDATA_DATABASE` | `dlt` | Display label used **only when creating** a new managed database (never to look one up) | | `schema` | `HOTDATA_SCHEMA` | `public` | Schema within the managed database | | `write_disposition` | `HOTDATA_WRITE_DISPOSITION` | `append` | Default write mode (see [Write modes](#write-modes)) | | `declared_tables` | `HOTDATA_DECLARED_TABLES` | — | All table names the pipeline will write (required for multi-table pipelines — see [Multiple tables](#multiple-tables)) | diff --git a/docs/runbook.md b/docs/runbook.md index b257b06..a3ad1bb 100644 --- a/docs/runbook.md +++ b/docs/runbook.md @@ -31,7 +31,10 @@ uv run hotdata-dlt-demo --workspace-id ``` - Requires `HOTDATA_API_KEY` to be set; pass your workspace id as the argument. + Requires `HOTDATA_API_KEY` to be set; pass your workspace id as the argument. On + the first run the demo creates a managed database and prints its id; pass that id + on later runs (`--database-id `) to load into the same database instead of + creating a new one (managed databases are addressed by id, not by name). ## Add a new pipeline diff --git a/docs/sql-client-spec.md b/docs/sql-client-spec.md index ed3b6a7..beb51c4 100644 --- a/docs/sql-client-spec.md +++ b/docs/sql-client-spec.md @@ -132,7 +132,7 @@ scoping (see §8): ```python def __init__(self, managed_database, schema, capabilities, config): super().__init__( - database_name=managed_database, # used only as the execute_sql(database=) scope + database_name=managed_database, # display label only; scoping resolves by id from config dataset_name=schema, # "public" -> drives default.public. staging_dataset_name=schema, # no staging; mirror dataset_name capabilities=capabilities, @@ -148,7 +148,7 @@ def __init__(self, managed_database, schema, capabilities, config): | `open_connection() -> HotdataClient` | Construct a `HotdataClient` from config, store on `self._client`, return it. No socket. | dlt opens the client when a dataset is materialized. | | `close_connection() -> None` | `self._client.close()`; clear it. | Dataset context exit. | | `native_connection -> HotdataClient` (property) | Return `self._client`. | Base `__getattr__` delegation; ibis backend. | -| `execute_query(query, *args, **kwargs) -> ContextManager[DBApiCursor]` | `@contextmanager`; run `self._client.execute_sql(sql, database=self.database_name)`, wrap the returned `pyarrow.Table` in `HotdataCursor`, `yield` it. Map SDK errors via `@raise_database_error`. | **Every read** — `Relation.to_sql()` → here. | +| `execute_query(query, *args, **kwargs) -> ContextManager[DBApiCursor]` | `@contextmanager`; run `self._client.execute_sql(sql)` (the database is resolved by id from the bound config), wrap the returned `pyarrow.Table` in `HotdataCursor`, `yield` it. Map SDK errors via `@raise_database_error`. | **Every read** — `Relation.to_sql()` → here. | | `execute_sql(query, *args, **kwargs) -> Sequence[Sequence] \| None` | `with self.execute_query(...) as c: return None if c.description is None else c.fetchall()`. | Base helpers, direct SQL. | | `begin_transaction() -> ContextManager[DBTransaction]` | No-op: `yield self`. Hotdata/DataFusion has no transactions (`supports_ddl_transactions=False`). | dlt may wrap ops in a txn. | | `_make_database_exception(ex) -> Exception` (static) | Map undefined-relation → `DatabaseUndefinedRelation`; transient → transient; else terminal. **Scan the whole `__cause__` chain**, not just `str(ex)`: the SDK's `classify_sdk_error` collapses the `ApiException` to `"400: Bad Request"`, so the engine's descriptive `"table … not found"` only appears deeper in the chain (in the underlying `hotdata.exceptions.BadRequestException`). | `@raise_database_error`. | @@ -180,7 +180,8 @@ class HotdataJobClient(JobClientBase, WithStateSync, WithSqlClient): # + WithS def sql_client(self) -> HotdataSqlClient: if self._sql_client is None: self._sql_client = HotdataSqlClient( - self.config.database_name, self.config.schema, self.capabilities, self.config + self.config.database_id or self.config.database_name, + self.config.schema, self.capabilities, self.config ) return self._sql_client ``` @@ -238,14 +239,13 @@ A single dlt "location" splits into **two independent mechanisms**: 1. **Table path — goes into the SQL.** `make_qualified_table_name("spans")` → `"default"."public"."spans"` because `catalog_name()="default"` and `dataset_name="public"`. -2. **Database scoping — goes into the request, not the SQL.** The managed database is passed as - `execute_sql(sql, database=self.database_name)`. Query scoping is by database **id** - (`_query_database_scoped(database_id=...)` → `X-Database-Id` header), so our `HotdataClient.execute_sql` - resolves **name → id** first via `resolve_managed_database(name).id` — exactly as `fetch_table` - already does. (This is what the CLI's `-d` flag supplies manually as an id; the SqlClient passes the - *name* and the id lookup lives in our client.) - -**Decision — mirror the write path:** address by the destination's managed `database_name` + the fixed +2. **Database scoping — goes into the request, not the SQL.** Query scoping is by database **id** + (`_query_database_scoped(database_id=...)` → `X-Database-Id` header). `HotdataClient.execute_sql` + resolves the run's database **by id** from the bound config (`database_id`, or the record created + this run) — never by name — exactly as `fetch_table` does. Managed databases are addressed by id + only; names are not unique and are not used to look one up. + +**Decision — mirror the write path:** address by the run's managed database **id** + the fixed `public` schema, exactly as writes do. Guarantees reads return what writes wrote, zero write-side change. (Consequence: dlt's pipeline `dataset_name` is not the addressing key — same as today, which is why load output shows `dataset None`.) Whether to make `dataset_name` idiomatic later is a joint diff --git a/pyproject.toml b/pyproject.toml index 602b50d..1ff2e06 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "hotdata-dlt-destination" -version = "0.10.0" +version = "0.11.0" description = "dlt destination for loading data into Hotdata managed databases." readme = "README.md" license = "MIT" @@ -19,7 +19,7 @@ dependencies = [ # decls (hotdata 0.8.0) and the per-load `key` on load_managed_table plus # create_managed_database(keys=) / add_managed_table(key=) (framework 0.8.0). "hotdata>=0.8.0,<0.9", - "hotdata-framework>=0.8.0,<0.9", + "hotdata-framework>=0.9.0,<0.10", "pandas>=2.0", "pyarrow>=14", ] diff --git a/scripts/load_test.py b/scripts/load_test.py index 6730612..bac3bb8 100644 --- a/scripts/load_test.py +++ b/scripts/load_test.py @@ -31,6 +31,7 @@ from concurrent.futures import ThreadPoolExecutor, as_completed from dataclasses import dataclass, field from datetime import UTC, datetime +from types import SimpleNamespace import pyarrow as pa import pyarrow.parquet as pq @@ -140,7 +141,7 @@ def run_database_load( workspace_id: str, api_base_url: str, do_query: bool, -) -> list[PhaseResult]: +) -> tuple[list[PhaseResult], str | None]: results: list[PhaseResult] = [] client = HotdataClient( @@ -150,23 +151,28 @@ def run_database_load( max_retries=3, retry_backoff_seconds=1.0, ) + # id-first: bind a run config so the created database is cached and every + # subsequent op addresses it by id (database_name is only the create label). + run_cfg = SimpleNamespace(database_id=None, database_name=db_name) + client.bind_run_cache(run_cfg) + created_id: str | None = None try: # --- create database --- t0 = time.perf_counter() try: - client.ensure_managed_database( - db_name, + db = client.ensure_managed_database( schema="public", tables=["events"], create_if_missing=True, ) + created_id = db.id results.append(PhaseResult(db_name, "create_db", time.perf_counter() - t0)) except Exception as exc: results.append( PhaseResult(db_name, "create_db", time.perf_counter() - t0, error=str(exc)) ) - return results + return results, created_id # --- generate + write parquet --- table = generate_table(rows) @@ -182,7 +188,7 @@ def run_database_load( results.append( PhaseResult(db_name, "write_parquet", time.perf_counter() - t0, error=str(exc)) ) - return results + return results, created_id # --- upload parquet --- t0 = time.perf_counter() @@ -191,7 +197,7 @@ def run_database_load( results.append(PhaseResult(db_name, "upload", time.perf_counter() - t0, rows=rows)) except Exception as exc: results.append(PhaseResult(db_name, "upload", time.perf_counter() - t0, error=str(exc))) - return results + return results, created_id finally: if parquet_path: with contextlib.suppress(OSError): @@ -200,17 +206,17 @@ def run_database_load( # --- load managed table --- t0 = time.perf_counter() try: - client.load_managed_table(db_name, "events", schema="public", upload_id=upload_id) + client.load_managed_table("events", schema="public", upload_id=upload_id) results.append(PhaseResult(db_name, "load", time.perf_counter() - t0, rows=rows)) except Exception as exc: results.append(PhaseResult(db_name, "load", time.perf_counter() - t0, error=str(exc))) - return results + return results, created_id # --- query via Arrow IPC --- if do_query: t0 = time.perf_counter() try: - result_table = client.fetch_table(database=db_name, schema="public", table="events") + result_table = client.fetch_table(schema="public", table="events") n = len(result_table) if result_table is not None else 0 results.append(PhaseResult(db_name, "query", time.perf_counter() - t0, rows=n)) except Exception as exc: @@ -221,7 +227,7 @@ def run_database_load( finally: client.close() - return results + return results, created_id # --------------------------------------------------------------------------- @@ -230,7 +236,7 @@ def run_database_load( def delete_databases( - db_names: list[str], + database_ids: list[str], *, api_key: str, workspace_id: str, @@ -240,13 +246,13 @@ def delete_databases( client = RuntimeClient(api_key, workspace_id, host=api_base_url.rstrip("/")) try: - for name in db_names: + for db_id in database_ids: try: - db = client.resolve_managed_database(name) - client.delete_managed_database(db.id) - print(f" deleted {name}") + # addressed by id (GET /databases/{id}) — no by-name lookup + client.delete_managed_database(db_id) + print(f" deleted {db_id}") except Exception as exc: - print(f" could not delete {name}: {exc}") + print(f" could not delete {db_id}: {exc}") finally: client.close() @@ -307,11 +313,14 @@ def main() -> None: } completed = 0 + created_ids: list[str] = [] for future in as_completed(futures): name = futures[future] completed += 1 try: - phase_results = future.result() + phase_results, created_id = future.result() + if created_id: + created_ids.append(created_id) except Exception as exc: print(f" [{completed:>3}/{args.databases}] {name} FATAL: {exc}") continue @@ -339,7 +348,7 @@ def main() -> None: if not args.no_cleanup: print("Cleaning up databases...") delete_databases( - db_names, + created_ids, api_key=api_key, workspace_id=workspace_id, api_base_url=api_base_url, diff --git a/scripts/roundtrip_demo.py b/scripts/roundtrip_demo.py index 8fdb8a4..07f3fa6 100644 --- a/scripts/roundtrip_demo.py +++ b/scripts/roundtrip_demo.py @@ -103,12 +103,34 @@ def read_without_dlt(workspace_id: str) -> None: def main() -> None: parser = argparse.ArgumentParser(description="dlt <-> Hotdata round trip demo.") parser.add_argument("--workspace-id", required=True, help="Hotdata workspace id") - workspace_id = parser.parse_args().workspace_id + parser.add_argument( + "--database-id", + default=None, + help="Existing managed database id (omit to create a new one by name)", + ) + args = parser.parse_args() + workspace_id = args.workspace_id + + # id-first: a managed database is addressed by id (names are not identifiers), + # so a read-back must know the id. Provision the database once here and pin its + # id for both the write and the read; pass --database-id to reuse an existing one. + database_id = args.database_id + if database_id is None: + from hotdata_framework.client import HotdataClient as RuntimeClient + + rc = RuntimeClient(os.environ["HOTDATA_API_KEY"], workspace_id, host=API_BASE_URL.rstrip("/")) + database_id = rc.create_managed_database( + description=DATABASE, schema="public", tables=["spans"] + ).id + rc.close() + print(f"created managed database {database_id} (pass --database-id {database_id} to reuse)") + pipeline = dlt.pipeline( pipeline_name="roundtrip_demo", destination=hotdata( credentials=HotdataCredentials(api_key=os.environ["HOTDATA_API_KEY"]), workspace_id=workspace_id, + database_id=database_id, database_name=DATABASE, declared_tables=["spans"], create_database_if_missing=True, diff --git a/src/hotdata_dlt_destination/cli.py b/src/hotdata_dlt_destination/cli.py index 90fab95..1d7e316 100644 --- a/src/hotdata_dlt_destination/cli.py +++ b/src/hotdata_dlt_destination/cli.py @@ -7,6 +7,7 @@ def main() -> None: config = HotdataDestinationConfig.from_env() print("hotdata-dlt-destination is configured") print(f"api_base_url={config.api_base_url}") + print(f"database_id={config.database_id or ''}") print(f"database_name={config.database_name}") print(f"schema={config.schema}") print(f"write_disposition={config.write_disposition}") diff --git a/src/hotdata_dlt_destination/config.py b/src/hotdata_dlt_destination/config.py index e5a9e5b..badfd9b 100644 --- a/src/hotdata_dlt_destination/config.py +++ b/src/hotdata_dlt_destination/config.py @@ -28,6 +28,7 @@ def _parse_backoff(value: str) -> float: class HotdataDestinationConfig: api_key: str database_name: str + database_id: str | None = None api_base_url: str = "https://api.hotdata.dev" schema: str = "public" write_disposition: str = "append" @@ -43,6 +44,7 @@ def from_env(cls) -> HotdataDestinationConfig: return cls( api_key=os.environ["HOTDATA_API_KEY"], database_name=os.environ.get("HOTDATA_DATABASE", "dlt"), + database_id=os.environ.get("HOTDATA_DATABASE_ID") or None, api_base_url=os.environ.get("HOTDATA_API_BASE_URL", "https://api.hotdata.dev"), schema=os.environ.get("HOTDATA_SCHEMA", "public"), write_disposition=os.environ.get("HOTDATA_WRITE_DISPOSITION", "append"), diff --git a/src/hotdata_dlt_destination/configuration.py b/src/hotdata_dlt_destination/configuration.py index 0ec2327..a66ca4f 100644 --- a/src/hotdata_dlt_destination/configuration.py +++ b/src/hotdata_dlt_destination/configuration.py @@ -32,8 +32,13 @@ class HotdataClientConfiguration(DestinationClientConfiguration): workspace_id: str | None = None """Hotdata workspace ID. Pass as a ``hotdata(workspace_id=...)`` param.""" api_base_url: str = "https://api.hotdata.dev" + database_id: str | None = None + """Id of the managed database to load into. This is how an existing database + is targeted — Hotdata database names are not unique, so the id (not the name) + is the identifier. Printed on first-run create; pin it to reuse the database.""" database_name: str = "dlt" - """Name of the managed database to load into.""" + """Display label for the managed database, used only when creating a new one + (never to look one up). Pin ``database_id`` to reuse an existing database.""" schema: str = "public" """Schema within the managed database.""" write_disposition: str = "append" diff --git a/src/hotdata_dlt_destination/factory.py b/src/hotdata_dlt_destination/factory.py index 3bbcde2..3f3de82 100644 --- a/src/hotdata_dlt_destination/factory.py +++ b/src/hotdata_dlt_destination/factory.py @@ -88,6 +88,7 @@ def __init__( self, credentials: HotdataCredentials | dict[str, t.Any] | str | None = None, workspace_id: str | None = None, + database_id: str | None = None, database_name: str = None, schema: str = None, write_disposition: str = None, @@ -123,6 +124,7 @@ def __init__( super().__init__( credentials=credentials, workspace_id=workspace_id, + database_id=database_id, database_name=database_name, schema=schema, write_disposition=write_disposition, diff --git a/src/hotdata_dlt_destination/hotdata_client.py b/src/hotdata_dlt_destination/hotdata_client.py index 38e0ffe..1c15d54 100644 --- a/src/hotdata_dlt_destination/hotdata_client.py +++ b/src/hotdata_dlt_destination/hotdata_client.py @@ -1,89 +1,134 @@ from __future__ import annotations import pyarrow as pa +from dlt.common import logger +from hotdata.api.databases_api import DatabasesApi from hotdata.arrow import ResultsApi as ArrowResultsApi -from hotdata_framework.databases import ManagedDatabase +from hotdata_framework.databases import ManagedDatabase, managed_database_from_detail from hotdata_framework.managed_client import ManagedDatabaseClient from hotdata_dlt_destination.errors import HotdataTerminalError +def _is_not_found(exc: Exception) -> bool: + """True when ``exc`` or anything in its ``__cause__`` chain wraps a 404. + + The framework raises the mapped error ``from`` the underlying ``ApiException``, + so today the 404 sits one level down -- but it walks the whole chain (mirroring + ``sql_client._chain_messages``) so a dropped-id 404 is still recognised if the + framework ever wraps ``get_database`` errors in an extra layer. + """ + current: BaseException | None = exc + for _ in range(6): + if current is None: + break + if getattr(current, "status", None) == 404: + return True + current = current.__cause__ + return False + + class HotdataClient(ManagedDatabaseClient): """Managed-database client used by the dlt destination. - Adds two things on top of the shared ``hotdata_framework`` client: - - * **Cross-run schema evolution** — when a later run requires a table the - database is missing, the table is declared in place via - ``add_managed_table`` (added empty, populated by the subsequent load); no - data is moved and existing tables, including dlt's ``_dlt_version`` / - ``_dlt_loads`` / ``_dlt_pipeline_state`` bookkeeping, are left untouched. - * **Collision-safe, resolve-once addressing** — a database name is resolved - to its record once per run (cached via :meth:`bind_run_cache`) and every - subsequent operation addresses the database by id. Resolution raises on an - ambiguous name instead of silently taking the first match. + Addressing is **id-first**: a Hotdata database is identified by its id, never + by name. The name (``description``) is only a display label supplied when a + database is created — it is never used to look one up, because Hotdata names + are not unique. Concretely: + + * **Bind an existing database by id.** When a ``database_id`` is configured, + the record is fetched once via ``GET /databases/{id}`` and every operation + addresses the database by that record. No listing, no name scan. + * **Create on first run.** With no ``database_id`` and ``create_if_missing``, + the database is created (labelled with ``database_name``) and its new id is + logged so it can be pinned via ``database_id`` for subsequent runs. + * **Cross-run schema evolution.** When binding an existing database, tables it + is missing are declared in place via ``add_managed_table`` (added empty, + populated by the subsequent load); no data is moved and existing tables, + including dlt's ``_dlt_version`` / ``_dlt_loads`` / ``_dlt_pipeline_state`` + bookkeeping, are left untouched. + + The resolved record is cached once per run (via :meth:`bind_run_cache`) so the + whole run reuses a single bind/create and create-scoped keys never issue a read + they are not permitted to make. """ - # Run-scoped store bound via bind_run_cache(); resolution is cached on it so - # the whole run reuses one resolved/created record. - _run_cache: object | None = None + # Run-scoped configuration bound via bind_run_cache(); the resolved database + # and its provenance are cached on it so the whole run reuses one record. + _config: object | None = None def bind_run_cache(self, cache: object) -> None: - """Bind a run-scoped store so a database resolves to its record once. + """Bind the run's shared configuration. - ``cache`` is any object that tolerates a ``_hotdata_resolved_db`` - attribute — in practice the shared ``HotdataClientConfiguration`` - instance, which every client built for a run points at. + ``cache`` is the shared ``HotdataClientConfiguration`` instance every + client built for a run points at. Its ``database_id`` / ``database_name`` + drive id-first resolution, and the resolved record is cached back on it + (``_hotdata_db``) so a single run resolves the database exactly once. """ - self._run_cache = cache + self._config = cache + + # --- run cache -------------------------------------------------------- + + def _cached_db(self) -> ManagedDatabase | None: + return getattr(self._config, "_hotdata_db", None) if self._config is not None else None - # --- resolution ------------------------------------------------------- + def _cache_db(self, db: ManagedDatabase | None, *, created: bool) -> None: + if self._config is not None: + self._config._hotdata_db = db + self._config._hotdata_db_created = created - def _collision_safe_resolve(self, name_or_id: str) -> ManagedDatabase: - """Resolve a name/id to its record, raising on an ambiguous name. + def _was_created(self) -> bool: + return bool(getattr(self._config, "_hotdata_db_created", False)) - Hotdata database names are not unique. Taking the first match can read, - write, or drop the wrong database, so a name that matches more than one - database raises instead. An id (matched exactly) is unambiguous. + # --- resolution (id-first, never by name) ----------------------------- + + def _bind_by_id(self, database_id: str) -> ManagedDatabase: + """Fetch a database record by id (``GET /databases/{id}``). + + Raises ``KeyError`` when the id does not exist (404); other API errors + propagate as ``HotdataTerminalError``/``HotdataTransientError``. """ - databases = self._request_with_retry(self._runtime.list_managed_databases) - by_name = [db for db in databases if db.description == name_or_id] - if len(by_name) > 1: - raise HotdataTerminalError( - f"Managed database name {name_or_id!r} is ambiguous: " - f"{len(by_name)} databases share it (ids: {sorted(db.id for db in by_name)}). " - "Address it by id to disambiguate." + try: + detail = self._request_with_retry( + lambda: DatabasesApi(self._runtime.api).get_database(database_id) ) - if by_name: - return by_name[0] - by_id = [db for db in databases if db.id == name_or_id] - if by_id: - return by_id[0] - raise KeyError(name_or_id) - - def _resolve(self, name_or_id: str) -> ManagedDatabase: - """Resolve once per run, then serve the cached (id-addressable) record.""" - cache = self._run_cache - if cache is not None: - cached = getattr(cache, "_hotdata_resolved_db", None) - if cached is not None and name_or_id in ( - cached.id, - getattr(cached, "description", None), - ): - return cached - db = self._collision_safe_resolve(name_or_id) - self._cache_db(db) - return db + except HotdataTerminalError as exc: + if _is_not_found(exc): + raise KeyError(database_id) from exc + raise + return managed_database_from_detail(detail) + + def _configured_database_id(self) -> str | None: + return getattr(self._config, "database_id", None) + + def _require_db(self) -> ManagedDatabase: + """Resolve the run's database for an operation (id-first). - def _cache_db(self, db: ManagedDatabase | None) -> None: - if self._run_cache is not None: - self._run_cache._hotdata_resolved_db = db + Reuses the run cache, else binds the configured ``database_id``. Raises + ``KeyError`` when no database has been resolved and none is configured — + there is deliberately no by-name fallback. + """ + db = self._cached_db() + if db is not None: + return db + database_id = self._configured_database_id() + if database_id: + db = self._bind_by_id(database_id) + self._cache_db(db, created=False) + return db + raise KeyError("no managed database resolved for this run (set database_id)") + + def resolved_database_id(self) -> str: + """Return the id of the run's managed database (bound by id or created). + + Raises ``KeyError`` when none is configured and none was created this run. + """ + return self._require_db().id # --- lifecycle -------------------------------------------------------- def ensure_managed_database( self, - name: str, *, schema: str, tables: list[str], @@ -93,71 +138,127 @@ def ensure_managed_database( # keys: table name -> key columns (enables delete/update/upsert on it) keys = keys or {} - try: - db = self._resolve(name) - except KeyError: - if not create_if_missing: - raise - db = self._request_with_retry( - lambda: self._runtime.create_managed_database( - description=name, schema=schema, tables=sorted(set(tables)), keys=keys - ) + db = self._cached_db() + created = self._was_created() + if db is None: + database_id = self._configured_database_id() + if database_id: + try: + db = self._bind_by_id(database_id) + except KeyError: + # A pinned id can't be recreated (ids are server-assigned), so + # on the create path a missing id is a clear terminal error, not + # a silent recreate. The probe path (create_if_missing=False, e.g. + # is_storage_initialized) still gets the KeyError -> "not there". + if create_if_missing: + raise HotdataTerminalError( + f"configured database_id {database_id!r} was not found " + "(it may have been dropped). A managed database cannot be " + "recreated with the same id -- unset database_id to create a " + "new one, or pin an existing id." + ) from None + raise + created = False + self._cache_db(db, created=False) + elif create_if_missing: + db = self._create(schema=schema, tables=tables, keys=keys) + created = True + self._cache_db(db, created=True) + else: + raise KeyError("no managed database resolved for this run (set database_id)") + + # A freshly created database already declared every table; only an + # existing (bound) database needs additive, in-place schema evolution. + if not created: + self._reconcile_tables(db, schema=schema, tables=tables, keys=keys) + return db + + def _create( + self, *, schema: str, tables: list[str], keys: dict[str, list[str]] + ) -> ManagedDatabase: + description = getattr(self._config, "database_name", None) + db = self._request_with_retry( + lambda: self._runtime.create_managed_database( + description=description, schema=schema, tables=sorted(set(tables)), keys=keys ) - self._cache_db(db) - return db + ) + # Logged at WARNING (dlt's default level) so the new id is always visible: + # without pinning it via database_id, the next run creates another database. + logger.warning( + "hotdata: created managed database %s (name=%r). Pin it for future runs by " + "setting database_id=%s (HOTDATA_DATABASE_ID / [destination.hotdata] database_id).", + db.id, + description, + db.id, + ) + return db + def _reconcile_tables( + self, + db: ManagedDatabase, + *, + schema: str, + tables: list[str], + keys: dict[str, list[str]], + ) -> None: existing = { managed_table.table for managed_table in self._request_with_retry( - lambda: self._runtime.list_managed_tables(db.id, schema=schema) + lambda: self._runtime.list_managed_tables(db, schema=schema) ) } # Declare any newly-required tables additively, in place, carrying their # key. dlt calls ``initialize_storage`` with the full table set before any # load job runs, so by load time this is normally a no-op. for table in sorted(set(tables) - existing): - self._add_managed_table(db.id, table, schema=schema, key=keys.get(table)) - return db + self._add_managed_table(db, table, schema=schema, key=keys.get(table)) def _add_managed_table( - self, database: str, table: str, *, schema: str, key: list[str] | None = None + self, + database: ManagedDatabase, + table: str, + *, + schema: str, + key: list[str] | None = None, ) -> None: self._request_with_retry( lambda: self._runtime.add_managed_table(database, table, schema=schema, key=key) ) - def drop_managed_database(self, name: str) -> None: - """Delete the managed database if it exists (used for dlt dev_mode / refresh).""" - try: - db = self._resolve(name) - except KeyError: - return - self._request_with_retry(lambda: self._runtime.delete_managed_database(db.id)) - self._cache_db(None) - - def resolve_managed_database(self, name: str) -> ManagedDatabase: - """Resolve a managed database by display name (or id) to its record. - - Raises ``KeyError`` when nothing matches and ``HotdataTerminalError`` when - the name is shared by more than one database. - """ - return self._resolve(name) - - def load_managed_table(self, database: str, table: str, **kwargs): - """Load parquet into a managed table, addressing the database by id.""" - db = self._resolve(database) - return super().load_managed_table(db.id, table, **kwargs) - - def execute_sql(self, sql: str, *, database: str) -> pa.Table: - """Run a SQL query scoped to ``database`` and return the result as Arrow. - - Resolves the managed database to its id (once per run), submits the query, - polls until the result is ready, and fetches it as a ``pyarrow.Table``. An - empty table is returned when the query produces no out-of-band result. + def drop_managed_database(self) -> None: + """Delete the run's managed database if it exists (used for dlt dev_mode / refresh).""" + db = self._cached_db() + if db is None: + database_id = self._configured_database_id() + if not database_id: + return + try: + db = self._bind_by_id(database_id) + except KeyError: + return + self._request_with_retry(lambda: self._runtime.delete_managed_database(db)) + self._cache_db(None, created=False) + + # --- operations (addressed by the resolved record) -------------------- + + def load_managed_table(self, table: str, **kwargs): + """Load parquet into a managed table via the resolved database record. + + Passing the resolved ``ManagedDatabase`` (not its id) lets a create-scoped + key load without a further read probe (framework passthrough).""" + db = self._require_db() + return super().load_managed_table(db, table, **kwargs) + + def execute_sql(self, sql: str) -> pa.Table: + """Run a SQL query scoped to the run's database and return Arrow. + + Submits the query, polls until the result is ready, and fetches it as a + ``pyarrow.Table``. An empty table is returned when the query produces no + out-of-band result. """ def operation() -> pa.Table: - db = self._resolve(database) + db = self._require_db() result_id = self._query_database_scoped(sql, database_id=db.id) if result_id is None: return pa.table({}) @@ -169,25 +270,16 @@ def operation() -> pa.Table: return self._request_with_retry(operation) - def list_managed_tables(self, database: str, *, schema: str) -> list: - """List the managed tables in ``database``/``schema`` (used by ``has_dataset``).""" - db = self._resolve(database) + def list_managed_tables(self, *, schema: str) -> list: + """List the managed tables in the run's database/``schema`` (used by ``has_dataset``).""" + db = self._require_db() return self._request_with_retry( - lambda: self._runtime.list_managed_tables(db.id, schema=schema) + lambda: self._runtime.list_managed_tables(db, schema=schema) ) - def table_is_synced(self, database: str, table: str, *, schema: str) -> bool: - db = self._resolve(database) - for managed_table in self._request_with_retry( - lambda: self._runtime.list_managed_tables(db.id, schema=schema) - ): - if managed_table.table == table: - return managed_table.synced - return False - - def fetch_table(self, *, database: str, schema: str, table: str) -> pa.Table | None: + def fetch_table(self, *, schema: str, table: str) -> pa.Table | None: def operation() -> pa.Table | None: - db = self._resolve(database) + db = self._require_db() if not self._table_is_synced_for(db, table, schema=schema): return None sql = f'SELECT * FROM "default"."{schema}"."{table}"' @@ -198,8 +290,12 @@ def operation() -> pa.Table | None: return self._request_with_retry(operation) + def fetch_table_rows(self, *, schema: str, table: str) -> list[dict]: + result = self.fetch_table(schema=schema, table=table) + return result.to_pylist() if result is not None else [] + def _table_is_synced_for(self, db: ManagedDatabase, table: str, *, schema: str) -> bool: - for managed_table in self._runtime.list_managed_tables(db.id, schema=schema): + for managed_table in self._runtime.list_managed_tables(db, schema=schema): if managed_table.table == table: return managed_table.synced return False diff --git a/src/hotdata_dlt_destination/ibis_backend.py b/src/hotdata_dlt_destination/ibis_backend.py index 9b73f8e..4565bb6 100644 --- a/src/hotdata_dlt_destination/ibis_backend.py +++ b/src/hotdata_dlt_destination/ibis_backend.py @@ -35,7 +35,7 @@ def ibis_connect(client: JobClientBase) -> Any: config = client.config with _hotdata_api(config) as api: - database_id = api.resolve_managed_database(config.database_name).id + database_id = api.resolved_database_id() return ibis.hotdata.connect( api_url=config.api_base_url, diff --git a/src/hotdata_dlt_destination/job_client.py b/src/hotdata_dlt_destination/job_client.py index 2a1347c..640631d 100644 --- a/src/hotdata_dlt_destination/job_client.py +++ b/src/hotdata_dlt_destination/job_client.py @@ -145,7 +145,6 @@ def _truncate_table( write_table_parquet(arrow_schema.empty_table(), parquet_path) upload_id = api.upload_parquet(parquet_path) api.load_managed_table( - config.database_name, table_name, schema=config.schema, upload_id=upload_id, @@ -167,7 +166,6 @@ def _upload_table( write_table_parquet(arrow_table, parquet_path) upload_id = api.upload_parquet(parquet_path) api.load_managed_table( - config.database_name, table_name, schema=config.schema, upload_id=upload_id, @@ -237,7 +235,6 @@ def _apply(api: HotdataClient, upload_table, mode: str) -> None: write_table_parquet(upload_table, parquet_path) upload_id = api.upload_parquet(parquet_path) api.load_managed_table( - contract.database_name, contract.table_name, schema=contract.schema, upload_id=upload_id, @@ -254,7 +251,6 @@ def _combine_and_replace( api: HotdataClient, combine_disposition: str, *, apply_hard_delete: bool = False ) -> None: existing = api.fetch_table( - database=contract.database_name, schema=contract.schema, table=contract.table_name, ) @@ -271,7 +267,6 @@ def _combine_and_replace( try: with _hotdata_api(self._config) as api: api.ensure_managed_database( - contract.database_name, schema=contract.schema, tables=_declared_tables( contract=contract, @@ -364,7 +359,7 @@ def sql_client(self) -> HotdataSqlClient: if self._sql_client is None: self._sql_client = HotdataSqlClient( - self.config.database_name, + self.config.database_id or self.config.database_name, self.config.schema, self.capabilities, self.config, @@ -398,7 +393,6 @@ def initialize_storage(self, truncate_tables: Iterable[str] = None) -> None: try: with _hotdata_api(self.config) as api: api.ensure_managed_database( - self.config.database_name, schema=self.config.schema, tables=all_tables, keys=keys, @@ -437,7 +431,6 @@ def is_storage_initialized(self) -> bool: with _hotdata_api(self.config) as api: try: api.ensure_managed_database( - self.config.database_name, schema=self.config.schema, tables=[], create_if_missing=False, @@ -452,7 +445,7 @@ def drop_storage(self) -> None: # recreates it on the next run. try: with _hotdata_api(self.config) as api: - api.drop_managed_database(self.config.database_name) + api.drop_managed_database() except HotdataTerminalError as exc: raise DestinationTerminalException(str(exc)) from exc @@ -549,7 +542,6 @@ def _fetch_internal_rows(self, table_name: str) -> list[dict[str, Any]]: with _hotdata_api(self.config) as api: try: table = api.fetch_table( - database=self.config.database_name, schema=self.config.schema, table=table_name, ) diff --git a/src/hotdata_dlt_destination/pipelines/demo.py b/src/hotdata_dlt_destination/pipelines/demo.py index b4fbf60..9b1c767 100644 --- a/src/hotdata_dlt_destination/pipelines/demo.py +++ b/src/hotdata_dlt_destination/pipelines/demo.py @@ -90,6 +90,12 @@ def macro_wide_resource(): def main() -> None: parser = argparse.ArgumentParser(description="Load FRED macro indicators into Hotdata.") parser.add_argument("--workspace-id", required=True, help="Hotdata workspace id") + parser.add_argument( + "--database-id", + default=None, + help="Existing managed database id to load into (printed on first-run create; " + "omit to create a new database by name)", + ) args = parser.parse_args() all_tables = ["macro_indicators_raw", "macro_wide"] @@ -98,6 +104,7 @@ def main() -> None: destination=hotdata( credentials=HotdataCredentials(api_key=os.environ["HOTDATA_API_KEY"]), workspace_id=args.workspace_id, + database_id=args.database_id, api_base_url=os.environ.get("HOTDATA_API_BASE_URL", "https://api.hotdata.dev"), write_disposition="replace", declared_tables=all_tables, diff --git a/src/hotdata_dlt_destination/pipelines/ibis_demo.py b/src/hotdata_dlt_destination/pipelines/ibis_demo.py index 17fc0d8..9b9ac63 100644 --- a/src/hotdata_dlt_destination/pipelines/ibis_demo.py +++ b/src/hotdata_dlt_destination/pipelines/ibis_demo.py @@ -84,13 +84,36 @@ def _read_with_ibis(pipeline: dlt.Pipeline) -> None: def main() -> None: parser = argparse.ArgumentParser(description="Load NYC taxi trips and read via ibis.") parser.add_argument("--workspace-id", required=True, help="Hotdata workspace id") + parser.add_argument( + "--database-id", + default=None, + help="Existing managed database id (omit to create a new one by name)", + ) args = parser.parse_args() + api_base_url = os.environ.get("HOTDATA_API_BASE_URL", "https://api.hotdata.dev") + + # id-first: the live ibis read binds the managed database by id (names are not + # identifiers). Provision it once and pin the id; pass --database-id to reuse. + database_id = args.database_id + if database_id is None: + from hotdata_framework.client import HotdataClient as RuntimeClient + + rc = RuntimeClient( + os.environ["HOTDATA_API_KEY"], args.workspace_id, host=api_base_url.rstrip("/") + ) + database_id = rc.create_managed_database( + description=DATABASE_NAME, schema=SCHEMA, tables=["trips"] + ).id + rc.close() + print(f"created managed database {database_id} (pass --database-id {database_id} to reuse)") + pipeline = dlt.pipeline( pipeline_name="ibis_demo", destination=hotdata( credentials=HotdataCredentials(api_key=os.environ["HOTDATA_API_KEY"]), workspace_id=args.workspace_id, - api_base_url=os.environ.get("HOTDATA_API_BASE_URL", "https://api.hotdata.dev"), + database_id=database_id, + api_base_url=api_base_url, write_disposition="replace", declared_tables=["trips"], database_name=DATABASE_NAME, diff --git a/src/hotdata_dlt_destination/pipelines/merge_demo.py b/src/hotdata_dlt_destination/pipelines/merge_demo.py index 04373cd..43ebde4 100644 --- a/src/hotdata_dlt_destination/pipelines/merge_demo.py +++ b/src/hotdata_dlt_destination/pipelines/merge_demo.py @@ -84,13 +84,37 @@ def _print_table(pipeline: dlt.Pipeline, label: str) -> None: def main() -> None: parser = argparse.ArgumentParser(description="Composite-key merge + hard_delete demo.") parser.add_argument("--workspace-id", required=True, help="Hotdata workspace id") + parser.add_argument( + "--database-id", + default=None, + help="Existing managed database id (omit to create a new one by name)", + ) args = parser.parse_args() + api_base_url = os.environ.get("HOTDATA_API_BASE_URL", "https://api.hotdata.dev") + + # id-first: address the managed database by id (names are not identifiers), so + # the before/after reads below can find it. Provision it once and pin the id; + # pass --database-id to reuse an existing database. + database_id = args.database_id + if database_id is None: + from hotdata_framework.client import HotdataClient as RuntimeClient + + rc = RuntimeClient( + os.environ["HOTDATA_API_KEY"], args.workspace_id, host=api_base_url.rstrip("/") + ) + database_id = rc.create_managed_database( + description=DATABASE, schema=SCHEMA, tables=["orders"] + ).id + rc.close() + print(f"created managed database {database_id} (pass --database-id {database_id} to reuse)") + pipeline = dlt.pipeline( pipeline_name="orders_merge", destination=hotdata( credentials=HotdataCredentials(api_key=os.environ["HOTDATA_API_KEY"]), workspace_id=args.workspace_id, - api_base_url=os.environ.get("HOTDATA_API_BASE_URL", "https://api.hotdata.dev"), + database_id=database_id, + api_base_url=api_base_url, declared_tables=["orders"], database_name=DATABASE, schema=SCHEMA, diff --git a/src/hotdata_dlt_destination/sql_client.py b/src/hotdata_dlt_destination/sql_client.py index 39edde0..1bec649 100644 --- a/src/hotdata_dlt_destination/sql_client.py +++ b/src/hotdata_dlt_destination/sql_client.py @@ -200,7 +200,7 @@ def execute_query(self, query: AnyStr, *args: Any, **kwargs: Any) -> Iterator[DB raise DatabaseTerminalException( NotImplementedError("HotdataSqlClient does not support parameterized queries") ) - table = self._client.execute_sql(query, database=self.database_name) + table = self._client.execute_sql(query) yield HotdataCursor(table) def execute_sql( @@ -246,7 +246,7 @@ def has_dataset(self) -> bool: # Check via the managed-DB API rather than the base's INFORMATION_SCHEMA.SCHEMATA # query (which binds %s params our query API can't take). try: - self._client.list_managed_tables(self.database_name, schema=self.dataset_name) + self._client.list_managed_tables(schema=self.dataset_name) return True except (KeyError, HotdataTerminalError, HotdataTransientError): return False diff --git a/tests/test_client.py b/tests/test_client.py index b9ba6a7..ecdde77 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -1,16 +1,22 @@ from types import SimpleNamespace import pytest +from hotdata.exceptions import ApiException, ForbiddenException from hotdata_dlt_destination.errors import HotdataTerminalError from hotdata_dlt_destination.hotdata_client import HotdataClient -def _db(db_id: str, name: str, conn: str = "conn") -> SimpleNamespace: +def _db(db_id: str, name: str = "dlt", conn: str = "conn") -> SimpleNamespace: return SimpleNamespace(id=db_id, description=name, default_connection_id=conn) -def _client(runtime, *, cache=None) -> HotdataClient: +def _cfg(*, database_id=None, database_name="dlt") -> SimpleNamespace: + # Stand-in for the shared HotdataClientConfiguration the run binds. + return SimpleNamespace(database_id=database_id, database_name=database_name) + + +def _client(runtime, *, config=None) -> HotdataClient: client = HotdataClient( api_key="k", workspace_id="ws", @@ -19,157 +25,241 @@ def _client(runtime, *, cache=None) -> HotdataClient: retry_backoff_seconds=0.0, ) client._runtime = runtime - if cache is not None: - client.bind_run_cache(cache) + if config is not None: + client.bind_run_cache(config) return client -# --- resolution: collision-safe, id-addressed, resolve-once --------------- +def _install_get_database(monkeypatch, registry: dict) -> None: + """Patch the id lookup (GET /databases/{id}); unknown ids 404.""" + class FakeDatabasesApi: + def __init__(self, api): + pass -def test_resolve_returns_single_match_by_name() -> None: - class FakeRuntime: - def list_managed_databases(self): - return [_db("db_1", "dlt")] + def get_database(self, database_id): + if database_id not in registry: + raise ApiException(status=404, reason="not found") + return registry[database_id] - def close(self): - return None + monkeypatch.setattr("hotdata_dlt_destination.hotdata_client.DatabasesApi", FakeDatabasesApi) + monkeypatch.setattr( + "hotdata_dlt_destination.hotdata_client.managed_database_from_detail", lambda d: d + ) - client = _client(FakeRuntime()) - assert client.resolve_managed_database("dlt").id == "db_1" - client.close() +class _Runtime: + """Fake runtime tracking managed-database lifecycle calls (no name lookups).""" -def test_resolve_by_id() -> None: - class FakeRuntime: - def list_managed_databases(self): - return [_db("db_1", "dlt")] + api = None - def close(self): - return None + def __init__(self, existing_tables=()) -> None: + self._existing_tables = list(existing_tables) + self.created: list[tuple] = [] + self.deleted: list[str] = [] + self.added: list[tuple] = [] + self.uploaded: list[str] = [] + self.loaded: list[tuple] = [] + self.table_lists = 0 - client = _client(FakeRuntime()) - assert client.resolve_managed_database("db_1").id == "db_1" + def list_managed_tables(self, database, *, schema): + self.table_lists += 1 + return [SimpleNamespace(table=t, synced=True) for t in self._existing_tables] + + def add_managed_table(self, database, table, *, schema, key=None): + self.added.append((getattr(database, "id", database), table, schema)) + self._existing_tables.append(table) + return SimpleNamespace(table=table, schema=schema) + + def create_managed_database(self, *, description, schema, tables, keys=None): + self.created.append((description, schema, list(tables))) + return _db("new_db", description) + + def delete_managed_database(self, db): + self.deleted.append(getattr(db, "id", db)) + + def upload_parquet(self, path): + self.uploaded.append(path) + return "up_1" + + def load_managed_table(self, database, table, *, schema, upload_id, mode="replace", key=None): + db_id = getattr(database, "id", database) + self.loaded.append((db_id, table, schema, upload_id)) + return SimpleNamespace( + connection_id="c", + schema_name=schema, + table_name=table, + row_count=1, + full_name=f"{db_id}.{schema}.{table}", + ) + + def close(self): + return None + + +# --- create on first run (no database_id) --------------------------------- + + +def test_create_when_no_id_configured() -> None: + rt = _Runtime() + client = _client(rt, config=_cfg(database_id=None, database_name="sales")) + db = client.ensure_managed_database(schema="public", tables=["orders"], create_if_missing=True) + # created, labelled by database_name; no read/list happened (nothing to reconcile) + assert rt.created == [("sales", "public", ["orders"])] + assert db.id == "new_db" + assert rt.table_lists == 0 client.close() -def test_resolve_missing_raises_keyerror() -> None: - class FakeRuntime: - def list_managed_databases(self): - return [] +def test_create_logs_new_id(monkeypatch) -> None: + messages: list[str] = [] + monkeypatch.setattr( + "hotdata_dlt_destination.hotdata_client.logger.warning", + lambda msg, *args: messages.append(msg % args if args else msg), + ) + rt = _Runtime() + client = _client(rt, config=_cfg(database_name="sales")) + client.ensure_managed_database(schema="public", tables=["orders"], create_if_missing=True) + # the id is surfaced for the user to pin via database_id + assert any("new_db" in m and "database_id" in m for m in messages) + client.close() - def close(self): - return None - client = _client(FakeRuntime()) +def test_no_id_and_no_create_raises_keyerror() -> None: + # is_storage_initialized() path: probe with create disabled and no id -> "not there". + rt = _Runtime() + client = _client(rt, config=_cfg(database_id=None)) with pytest.raises(KeyError): - client.resolve_managed_database("dlt") + client.ensure_managed_database(schema="public", tables=["orders"], create_if_missing=False) + assert rt.created == [] client.close() -def test_resolve_raises_on_ambiguous_name() -> None: - # Hotdata names are not unique: >1 match must raise, never silently pick one. - class FakeRuntime: - def list_managed_databases(self): - return [_db("db_1", "dlt"), _db("db_2", "dlt")] +# --- bind an existing database by id -------------------------------------- - def close(self): - return None - client = _client(FakeRuntime()) - with pytest.raises(HotdataTerminalError, match="ambiguous"): - client.resolve_managed_database("dlt") +def test_bind_existing_by_id(monkeypatch) -> None: + _install_get_database(monkeypatch, {"db_1": _db("db_1", "sales")}) + rt = _Runtime(existing_tables=["orders", "customers"]) + client = _client(rt, config=_cfg(database_id="db_1")) + db = client.ensure_managed_database(schema="public", tables=["orders"], create_if_missing=True) + # bound by id; not recreated + assert db.id == "db_1" + assert rt.created == [] client.close() -def test_resolves_once_and_reuses_run_cache() -> None: - # The name is resolved once per run; a second client sharing the cache reuses it. - class FakeRuntime: - def __init__(self): - self.list_calls = 0 +def test_bind_missing_id_raises_keyerror(monkeypatch) -> None: + _install_get_database(monkeypatch, {}) # id does not exist -> 404 + rt = _Runtime() + client = _client(rt, config=_cfg(database_id="db_missing")) + with pytest.raises(KeyError): + client.ensure_managed_database(schema="public", tables=["orders"], create_if_missing=False) + assert rt.created == [] + client.close() - def list_managed_databases(self): - self.list_calls += 1 - return [_db("db_1", "dlt")] - def list_managed_tables(self, database, *, schema): - assert database == "db_1" # addressed by id after the first resolve - return [] +def test_is_not_found_walks_cause_chain() -> None: + # A 404 buried under extra wrapping layers is still recognised, so the clear + # "database_id was dropped" message survives even if the framework nests errors. + from hotdata_dlt_destination.hotdata_client import _is_not_found + + inner = ApiException(status=404, reason="not found") + mid = HotdataTerminalError("wrapped") + mid.__cause__ = inner + outer = HotdataTerminalError("outer") + outer.__cause__ = mid + assert _is_not_found(outer) is True + + non_404 = HotdataTerminalError("x") + non_404.__cause__ = ApiException(status=500, reason="boom") + assert _is_not_found(non_404) is False + + +def test_bind_missing_id_with_create_raises_clear_error(monkeypatch) -> None: + # dev_mode/refresh drops the pinned db, then a create-path ensure finds the id + # gone. It can't be recreated (ids are server-assigned) -> clear terminal error, + # not a raw KeyError. + _install_get_database(monkeypatch, {}) # pinned id no longer exists -> 404 + rt = _Runtime() + client = _client(rt, config=_cfg(database_id="db_dropped")) + with pytest.raises(HotdataTerminalError, match="db_dropped"): + client.ensure_managed_database(schema="public", tables=["orders"], create_if_missing=True) + assert rt.created == [] # never silently recreated + client.close() - def close(self): - return None - rt = FakeRuntime() - cache = SimpleNamespace() - client = _client(rt, cache=cache) - client.resolve_managed_database("dlt") - client.list_managed_tables("dlt", schema="public") +def test_bind_by_id_evolves_schema_in_place(monkeypatch) -> None: + _install_get_database(monkeypatch, {"db_1": _db("db_1", "sales")}) + rt = _Runtime(existing_tables=["orders"]) + client = _client(rt, config=_cfg(database_id="db_1")) + client.ensure_managed_database( + schema="public", tables=["orders", "customers"], create_if_missing=True + ) + # the missing table is declared in place, addressed by the resolved id + assert rt.added == [("db_1", "customers", "public")] + assert rt.created == [] + assert rt.deleted == [] + client.close() - client2 = _client(rt, cache=cache) - assert client2.resolve_managed_database("dlt").id == "db_1" - assert rt.list_calls == 1 - assert cache._hotdata_resolved_db.id == "db_1" - client.close() - client2.close() +# --- resolve-once / run cache (populated by id or create, never by name) -- -# --- upload / load: addressed by id --------------------------------------- +def test_resolves_once_and_reuses_run_cache(monkeypatch) -> None: + reads = {"n": 0} + class FakeDatabasesApi: + def __init__(self, api): + pass -def test_upload_and_load_managed_table_addresses_by_id() -> None: - class FakeRuntime: - def __init__(self) -> None: - self.upload_calls = 0 - self.load_database = None + def get_database(self, database_id): + reads["n"] += 1 + return _db("db_1", "sales") - def list_managed_databases(self): - return [_db("db_dlt", "dlt")] + monkeypatch.setattr("hotdata_dlt_destination.hotdata_client.DatabasesApi", FakeDatabasesApi) + monkeypatch.setattr( + "hotdata_dlt_destination.hotdata_client.managed_database_from_detail", lambda d: d + ) - def upload_parquet(self, path: str) -> str: - self.upload_calls += 1 - assert path.endswith(".parquet") - return "upload_1" + cfg = _cfg(database_id="db_1") + rt = _Runtime(existing_tables=["orders"]) + client = _client(rt, config=cfg) + client.ensure_managed_database(schema="public", tables=["orders"], create_if_missing=True) + # a second client sharing the run config reuses the cached record + client2 = _client(_Runtime(existing_tables=["orders"]), config=cfg) + assert client2.resolved_database_id() == "db_1" + assert reads["n"] == 1 # bound exactly once for the run + assert cfg._hotdata_db.id == "db_1" + client.close() + client2.close() - def load_managed_table( - self, database, table, *, schema, upload_id, mode="replace", key=None - ): - self.load_database = database - return SimpleNamespace( - connection_id="conn_1", - schema_name=schema, - table_name=table, - row_count=1, - full_name=f"{database}.{schema}.{table}", - ) - def close(self) -> None: - return None +# --- operations address the resolved record ------------------------------- - rt = FakeRuntime() - client = _client(rt) +def test_load_addresses_by_resolved_record(monkeypatch) -> None: + _install_get_database(monkeypatch, {"db_dlt": _db("db_dlt", "dlt")}) + rt = _Runtime() + client = _client(rt, config=_cfg(database_id="db_dlt")) upload_id = client.upload_parquet("/tmp/batch.parquet") - loaded = client.load_managed_table("dlt", "orders", schema="public", upload_id=upload_id) - - assert upload_id == "upload_1" - # the load is addressed by the resolved id, not the display name - assert rt.load_database == "db_dlt" + loaded = client.load_managed_table("orders", schema="public", upload_id=upload_id) + assert upload_id == "up_1" + assert rt.loaded == [("db_dlt", "orders", "public", "up_1")] assert loaded.full_name == "db_dlt.public.orders" - assert rt.upload_calls == 1 client.close() -# --- fetch_table / execute_sql: carry the database scope ------------------ +def test_list_managed_tables_by_id(monkeypatch) -> None: + _install_get_database(monkeypatch, {"db_1": _db("db_1")}) + rt = _Runtime(existing_tables=["orders"]) + client = _client(rt, config=_cfg(database_id="db_1")) + tables = client.list_managed_tables(schema="public") + assert [t.table for t in tables] == ["orders"] + client.close() def _patch_query_apis(monkeypatch, arrow_table, *, arrow_in_hotdata_client: bool) -> dict: - """Patch the query/result APIs so no real HTTP happens. - - ``fetch_table`` resolves ArrowResultsApi from hotdata_framework.managed_client; - ``execute_sql`` resolves it from hotdata_dlt_destination.hotdata_client. - Returns a recorder of the ``x_database_id`` scopes each read carried. - """ from hotdata.models.query_response import QueryResponse as _QR scopes: dict[str, list] = {"result": [], "arrow": []} @@ -217,222 +307,120 @@ def get_result_arrow(self, result_id, *, x_database_id): return scopes -def test_fetch_table_carries_database_scope(monkeypatch) -> None: - # Hosted result endpoints reject requests without the database scope; - # fetch_table (merge / state-sync read-back) must carry it on the result - # poll and the Arrow fetch, addressing by the resolved id. - import pyarrow as pa - - scopes = _patch_query_apis(monkeypatch, pa.table({"id": [1]}), arrow_in_hotdata_client=False) - - class FakeRuntime: - api = None - - def list_managed_databases(self): - return [_db("db_42", "dlt")] - - def list_managed_tables(self, database, *, schema): - assert database == "db_42" - return [SimpleNamespace(table="orders", synced=True)] - - def close(self): - return None - - client = _client(FakeRuntime()) - table = client.fetch_table(database="dlt", schema="public", table="orders") - assert table is not None and table.num_rows == 1 - assert scopes["result"] == ["db_42"] - assert scopes["arrow"] == ["db_42"] - client.close() - - def test_execute_sql_carries_database_scope(monkeypatch) -> None: import pyarrow as pa + _install_get_database(monkeypatch, {"db_99": _db("db_99")}) scopes = _patch_query_apis(monkeypatch, pa.table({"id": [1, 2]}), arrow_in_hotdata_client=True) - class FakeRuntime: - api = None - - def list_managed_databases(self): - return [_db("db_99", "dlt")] - - def close(self): - return None - - client = _client(FakeRuntime()) - table = client.execute_sql('SELECT * FROM "default"."public"."spans"', database="dlt") + client = _client(_Runtime(), config=_cfg(database_id="db_99")) + table = client.execute_sql('SELECT * FROM "default"."public"."spans"') assert table.num_rows == 2 assert scopes["result"] == ["db_99"] assert scopes["arrow"] == ["db_99"] client.close() -def test_fetch_table_rows_skips_unsynced_tables() -> None: - class FakeRuntime: - def list_managed_databases(self): - return [_db("db_1", "dlt")] - - def list_managed_tables(self, database, *, schema): - assert database == "db_1" - return [SimpleNamespace(table="orders", synced=False)] - - def close(self) -> None: - return None +def test_fetch_table_carries_database_scope(monkeypatch) -> None: + import pyarrow as pa - client = _client(FakeRuntime()) - rows = client.fetch_table_rows(database="dlt", schema="public", table="orders") - assert rows == [] + _install_get_database(monkeypatch, {"db_42": _db("db_42")}) + scopes = _patch_query_apis(monkeypatch, pa.table({"id": [1]}), arrow_in_hotdata_client=False) + rt = _Runtime(existing_tables=["orders"]) + client = _client(rt, config=_cfg(database_id="db_42")) + table = client.fetch_table(schema="public", table="orders") + assert table is not None and table.num_rows == 1 + assert scopes["result"] == ["db_42"] + assert scopes["arrow"] == ["db_42"] client.close() -def test_fetch_table_rows_returns_empty_when_table_missing() -> None: - class FakeRuntime: - def list_managed_databases(self): - return [_db("db_1", "dlt")] +def test_fetch_table_rows_skips_unsynced_tables(monkeypatch) -> None: + _install_get_database(monkeypatch, {"db_1": _db("db_1")}) + class _Unsynced(_Runtime): def list_managed_tables(self, database, *, schema): - return [] - - def close(self) -> None: - return None + return [SimpleNamespace(table="orders", synced=False)] - client = _client(FakeRuntime()) - rows = client.fetch_table_rows(database="dlt", schema="public", table="orders") - assert rows == [] + client = _client(_Unsynced(), config=_cfg(database_id="db_1")) + assert client.fetch_table_rows(schema="public", table="orders") == [] client.close() -# --- ensure_managed_database / drop_managed_database (schema evolution) --- - - -class _EvoRuntime: - """Fake runtime tracking the managed-database lifecycle calls.""" - - def __init__(self, existing_db, existing_tables) -> None: - self._existing_db = existing_db # _db(...) or None - self._existing_tables = list(existing_tables) - self.created: list[tuple] = [] - self.deleted: list[str] = [] - self.added: list[tuple] = [] - self.uploaded: list[str] = [] - self.loaded: list[tuple] = [] - - def list_managed_databases(self): - return [self._existing_db] if self._existing_db is not None else [] - - def list_managed_tables(self, database, *, schema): - return [SimpleNamespace(table=t) for t in self._existing_tables] - - def add_managed_table(self, database, table, *, schema, key=None): - self.added.append((database, table, schema)) - self._existing_tables.append(table) - return SimpleNamespace(table=table, schema=schema) - - def create_managed_database(self, *, description, schema, tables, keys=None): - self.created.append((description, schema, list(tables))) - return _db("new_db", description) - - def delete_managed_database(self, db_id): - self.deleted.append(db_id) - - def upload_parquet(self, path): - self.uploaded.append(path) - return "up_1" +# --- drop ----------------------------------------------------------------- - def load_managed_table(self, database, table, *, schema, upload_id, mode="replace", key=None): - self.loaded.append((database, table, schema, upload_id)) - return SimpleNamespace() - def close(self): - return None +def test_drop_deletes_by_id(monkeypatch) -> None: + _install_get_database(monkeypatch, {"db_1": _db("db_1")}) + rt = _Runtime() + cfg = _cfg(database_id="db_1") + client = _client(rt, config=cfg) + client.drop_managed_database() + assert rt.deleted == ["db_1"] + assert cfg._hotdata_db is None + client.close() -def test_ensure_creates_when_missing() -> None: - rt = _EvoRuntime(existing_db=None, existing_tables=[]) - client = _client(rt) - client.ensure_managed_database("db", schema="public", tables=["orders"], create_if_missing=True) - assert rt.created == [("db", "public", ["orders"])] +def test_drop_noop_without_id() -> None: + rt = _Runtime() + client = _client(rt, config=_cfg(database_id=None)) + client.drop_managed_database() assert rt.deleted == [] client.close() -def test_ensure_raises_when_missing_and_no_create() -> None: - rt = _EvoRuntime(existing_db=None, existing_tables=[]) - client = _client(rt) - with pytest.raises(KeyError): - client.ensure_managed_database( - "db", schema="public", tables=["orders"], create_if_missing=False - ) - client.close() +# --- #55: create/upload/query-scoped keys (forbidden reads) bootstrap ------ +# Under id-first there is no id to look up on a first run, so the create path +# never issues a read the key isn't allowed to make -- create just succeeds. -def test_ensure_noop_when_all_tables_present() -> None: - rt = _EvoRuntime(existing_db=_db("db_1", "db"), existing_tables=["orders", "customers"]) - client = _client(rt) - client.ensure_managed_database("db", schema="public", tables=["orders"], create_if_missing=True) - assert rt.deleted == [] - assert rt.created == [] - assert rt.added == [] - client.close() +class _CreateScopedRuntime(_Runtime): + """A create-scoped key: any read (get_database / list) is forbidden.""" + def list_managed_tables(self, database, *, schema): + raise ForbiddenException(status=403, reason="ACCESS_DENIED") -def test_ensure_adds_missing_table_without_recreate() -> None: - rt = _EvoRuntime(existing_db=_db("db_1", "db"), existing_tables=["orders"]) - client = _client(rt) - client.ensure_managed_database( - "db", schema="public", tables=["orders", "customers"], create_if_missing=True - ) +def test_create_scoped_key_bootstraps_without_read(monkeypatch) -> None: + # get_database must never be called on the create path. + called = {"get": 0} - # The missing table is declared in place, addressed by the resolved id; the - # database is never deleted or recreated and no data is moved. - assert rt.added == [("db_1", "customers", "public")] - assert rt.deleted == [] - assert rt.created == [] - assert rt.uploaded == [] - assert rt.loaded == [] - client.close() + class FakeDatabasesApi: + def __init__(self, api): + pass + def get_database(self, database_id): + called["get"] += 1 + raise ForbiddenException(status=403, reason="ACCESS_DENIED") -def test_ensure_raises_on_ambiguous_name() -> None: - class _Ambiguous(_EvoRuntime): - def list_managed_databases(self): - return [_db("db_1", "db"), _db("db_2", "db")] + monkeypatch.setattr("hotdata_dlt_destination.hotdata_client.DatabasesApi", FakeDatabasesApi) - rt = _Ambiguous(existing_db=_db("db_1", "db"), existing_tables=[]) - client = _client(rt) - with pytest.raises(HotdataTerminalError, match="ambiguous"): - client.ensure_managed_database( - "db", schema="public", tables=["orders"], create_if_missing=True - ) - # never created a duplicate on the ambiguity - assert rt.created == [] + rt = _CreateScopedRuntime() + cfg = _cfg(database_id=None, database_name="dlt") + client = _client(rt, config=cfg) + db = client.ensure_managed_database(schema="public", tables=["orders"], create_if_missing=True) + assert rt.created == [("dlt", "public", ["orders"])] + assert db.id == "new_db" + assert called["get"] == 0 # never attempted a forbidden read + assert cfg._hotdata_db.id == "new_db" client.close() -def test_drop_managed_database_deletes_by_id_when_present() -> None: - rt = _EvoRuntime(existing_db=_db("db_1", "db"), existing_tables=[]) - client = _client(rt) - client.drop_managed_database("db") - assert rt.deleted == ["db_1"] - client.close() - +def test_create_scoped_load_reuses_cache_without_read(monkeypatch) -> None: + class FakeDatabasesApi: + def __init__(self, api): + pass -def test_drop_managed_database_noop_when_absent() -> None: - rt = _EvoRuntime(existing_db=None, existing_tables=[]) - client = _client(rt) - client.drop_managed_database("db") - assert rt.deleted == [] - client.close() + def get_database(self, database_id): + raise ForbiddenException(status=403, reason="ACCESS_DENIED") + monkeypatch.setattr("hotdata_dlt_destination.hotdata_client.DatabasesApi", FakeDatabasesApi) -def test_drop_clears_run_cache() -> None: - rt = _EvoRuntime(existing_db=_db("db_1", "db"), existing_tables=[]) - cache = SimpleNamespace() - client = _client(rt, cache=cache) - client.resolve_managed_database("db") - assert cache._hotdata_resolved_db.id == "db_1" - client.drop_managed_database("db") - assert cache._hotdata_resolved_db is None + rt = _CreateScopedRuntime() + cfg = _cfg(database_id=None, database_name="dlt") + client = _client(rt, config=cfg) + client.ensure_managed_database(schema="public", tables=["orders"], create_if_missing=True) + # a load in the same run resolves from cache and hands the record straight through + client.load_managed_table("orders", schema="public", upload_id="u1") + assert rt.loaded == [("new_db", "orders", "public", "u1")] client.close() diff --git a/tests/test_e2e_inmemory.py b/tests/test_e2e_inmemory.py index 6f41949..798b22e 100644 --- a/tests/test_e2e_inmemory.py +++ b/tests/test_e2e_inmemory.py @@ -19,6 +19,7 @@ import pyarrow.parquet as pq import pytest from dlt.common.schema import Schema +from hotdata.exceptions import ApiException from hotdata.models.query_response import QueryResponse import hotdata_dlt_destination.job_client as jc @@ -49,19 +50,18 @@ def __init__(self) -> None: def close(self) -> None: pass - def resolve_managed_database(self, name_or_id): - name = self.id_to_name.get(name_or_id, name_or_id) - if name not in self.name_to_id: - raise KeyError(name) - return SimpleNamespace(id=self.name_to_id[name], default_connection_id="conn") - - def list_managed_databases(self): - return [ - SimpleNamespace(id=db_id, description=name, default_connection_id="conn") - for name, db_id in self.name_to_id.items() - ] + def get_database(self, database_id): + # id-first bind: GET /databases/{id}; unknown id 404s. + if database_id not in self.id_to_name: + raise ApiException(status=404, reason="not found") + return SimpleNamespace( + id=database_id, + description=self.id_to_name[database_id], + default_connection_id="conn", + ) def list_managed_tables(self, database, *, schema=None): + database = getattr(database, "id", database) # accept a resolved ManagedDatabase or id/name name = self.id_to_name.get(database, database) return [ SimpleNamespace(table=t, var_schema="public", synced=(name, "public", t) in self.tables) @@ -79,6 +79,7 @@ def create_managed_database(self, *, description, schema, tables, keys=None, exp return SimpleNamespace(id=db_id, description=description, default_connection_id="conn") def delete_managed_database(self, name_or_id): + name_or_id = getattr(name_or_id, "id", name_or_id) name = self.id_to_name.get(name_or_id, name_or_id) db_id = self.name_to_id.pop(name, None) self.id_to_name.pop(db_id, None) @@ -87,6 +88,7 @@ def delete_managed_database(self, name_or_id): del self.tables[key] def add_managed_table(self, database, table, *, schema, key=None): + database = getattr(database, "id", database) name = self.id_to_name.get(database, database) self.declared.setdefault(name, set()).add(table) self.keys[(name, schema, table)] = list(key or []) @@ -101,6 +103,7 @@ def upload_parquet(self, path): def load_managed_table(self, database, table, *, schema, upload_id, mode="replace", key=None): import pyarrow as _pa + database = getattr(database, "id", database) name = self.id_to_name.get(database, database) k = (name, schema, table) self.load_counts[k] = self.load_counts.get(k, 0) + 1 @@ -230,6 +233,16 @@ def __init__(self, **kwargs): self._runtime = _ACTIVE["backend"] +class _FakeDatabasesApi: + """Routes the id-first bind (GET /databases/{id}) to the active backend.""" + + def __init__(self, api): + pass + + def get_database(self, database_id): + return _ACTIVE["backend"].get_database(database_id) + + @pytest.fixture def backend(monkeypatch): be = InMemoryBackend() @@ -237,10 +250,15 @@ def backend(monkeypatch): monkeypatch.setattr(mc, "QueryApi", _FakeQueryApi) monkeypatch.setattr(mc, "ResultsApi", _FakeResultsApi) monkeypatch.setattr(mc, "ArrowResultsApi", _FakeArrowResultsApi) - # hotdata_client.py also references ArrowResultsApi (for execute_sql). + # hotdata_client.py also references ArrowResultsApi (for execute_sql) and + # DatabasesApi (for the id-first bind). monkeypatch.setattr( "hotdata_dlt_destination.hotdata_client.ArrowResultsApi", _FakeArrowResultsApi ) + monkeypatch.setattr("hotdata_dlt_destination.hotdata_client.DatabasesApi", _FakeDatabasesApi) + monkeypatch.setattr( + "hotdata_dlt_destination.hotdata_client.managed_database_from_detail", lambda d: d + ) monkeypatch.setattr(jc, "HotdataClient", _E2EClient) monkeypatch.setattr(sc, "HotdataClient", _E2EClient) yield be @@ -248,9 +266,19 @@ def backend(monkeypatch): def _dest(database_name, declared_tables, write_disposition="append"): + # id-first: a managed database is addressed by id. Model the real workflow — + # the database exists (created once) and the pipeline pins its id — by + # creating it up front and binding by id. `database_name` stays the label. + be = _ACTIVE["backend"] + database_id = be.name_to_id.get(database_name) + if database_id is None: + database_id = be.create_managed_database( + description=database_name, schema="public", tables=[] + ).id return hotdata( credentials=HotdataCredentials(api_key="test"), workspace_id="ws_test", + database_id=database_id, database_name=database_name, declared_tables=declared_tables, write_disposition=write_disposition, @@ -276,6 +304,32 @@ def orders(): assert backend.rows("e2e_basic", "_dlt_version") is not None +def test_auto_create_when_no_database_id(backend, tmp_path): + # First run with no database_id: the pipeline creates the managed database by + # its name (label) and addresses it by the returned id for the rest of the run. + @dlt.resource(name="orders", write_disposition="replace") + def orders(): + yield [{"id": 1, "amount": 10}] + + dlt.pipeline( + pipeline_name="p_autocreate", + destination=hotdata( + credentials=HotdataCredentials(api_key="test"), + workspace_id="ws_test", + database_name="e2e_autocreate", + declared_tables=["orders"], + write_disposition="replace", + ), + dataset_name="public", + pipelines_dir=str(tmp_path), + ).run(orders()) + + # Exactly one database was created, under the given label. + assert set(backend.name_to_id) == {"e2e_autocreate"} + rows = backend.rows("e2e_autocreate", "orders") + assert rows is not None and len(rows) == 1 + + def test_replace_multi_file_pipeline_keeps_all_rows(backend, tmp_path, monkeypatch): # Regression for multi-file replace data loss: with a small # DATA_WRITER__FILE_MAX_BYTES (the ingest worker sets 200MB in production), @@ -371,6 +425,7 @@ def events(updated=dlt.sources.incremental("updated")): # noqa: B008 (dlt idio cfg = HotdataClientConfiguration( credentials=HotdataCredentials(api_key="test"), workspace_id="ws_test", + database_id=backend.name_to_id["e2e_state"], database_name="e2e_state", schema="public", ) diff --git a/tests/test_factory.py b/tests/test_factory.py index 61859e8..fe04288 100644 --- a/tests/test_factory.py +++ b/tests/test_factory.py @@ -71,6 +71,20 @@ def test_workspace_id_is_none_when_not_passed(clean_env) -> None: assert cfg.workspace_id is None +def test_database_id_param_flows_to_config(clean_env) -> None: + # An existing database is targeted by id (names are not unique identifiers). + clean_env.setenv("HOTDATA_API_KEY", "sk_env") + cfg = _resolve(hotdata(workspace_id="ws", database_id="db_abc", declared_tables=["t"])) + assert cfg.database_id == "db_abc" + + +def test_database_id_is_none_when_not_passed(clean_env) -> None: + # No id means "create by name on first run"; the id is None until then. + clean_env.setenv("HOTDATA_API_KEY", "sk_env") + cfg = _resolve(hotdata(workspace_id="ws", database_name="d", declared_tables=["t"])) + assert cfg.database_id is None + + def test_legacy_workspace_in_credentials_dict_is_hoisted_without_mutating(clean_env) -> None: creds = {"api_key": "k", "workspace_id": "ws_dict"} with pytest.warns(DeprecationWarning): diff --git a/tests/test_job_client.py b/tests/test_job_client.py index 92b3860..a834fca 100644 --- a/tests/test_job_client.py +++ b/tests/test_job_client.py @@ -34,17 +34,17 @@ def __init__(self, **_kwargs: object) -> None: def bind_run_cache(self, cache: object) -> None: return None - def ensure_managed_database(self, name, *, schema, tables, keys=None, create_if_missing): + def ensure_managed_database(self, *, schema, tables, keys=None, create_if_missing): return SimpleNamespace(id="db_1") - def fetch_table(self, *, database, schema, table): + def fetch_table(self, *, schema, table): return store.get(table) def upload_parquet(self, path: str) -> str: self._pending = pq.read_table(path) return "upload_1" - def load_managed_table(self, database, table, *, schema, upload_id, mode="replace", key=None): + def load_managed_table(self, table, *, schema, upload_id, mode="replace", key=None): assert self._pending is not None # Mode-faithful, like the server: append accumulates, replace overwrites. existing = store.get(table) @@ -54,7 +54,7 @@ def load_managed_table(self, database, table, *, schema, upload_id, mode="replac ) else: store[table] = self._pending - return SimpleNamespace(full_name=f"{database}.{schema}.{table}") + return SimpleNamespace(full_name=f"db_1.{schema}.{table}") def close(self) -> None: return None @@ -202,18 +202,18 @@ def __init__(self, **_kwargs: object) -> None: def bind_run_cache(self, cache: object) -> None: return None - def ensure_managed_database(self, name, *, schema, tables, keys=None, create_if_missing): + def ensure_managed_database(self, *, schema, tables, keys=None, create_if_missing): calls["keys"] = keys return SimpleNamespace(id="db_1") - def fetch_table(self, *, database, schema, table): + def fetch_table(self, *, schema, table): calls["fetches"] = calls.get("fetches", 0) + 1 def upload_parquet(self, path: str) -> str: self._pending = pq.read_table(path) return "upload_1" - def load_managed_table(self, database, table, *, schema, upload_id, mode="replace", key=None): + def load_managed_table(self, table, *, schema, upload_id, mode="replace", key=None): calls.setdefault("modes", []).append(mode) calls["mode"] = mode calls["load_key"] = key @@ -225,7 +225,7 @@ def load_managed_table(self, database, table, *, schema, upload_id, mode="replac ) if reject_mode is not None and mode == reject_mode: raise HotdataTerminalError(f"{table}: no declared key; required for mode={mode}") - return SimpleNamespace(full_name=f"{database}.{schema}.{table}") + return SimpleNamespace(full_name=f"db_1.{schema}.{table}") def close(self) -> None: return None @@ -496,8 +496,8 @@ def __init__(self, **_kwargs: object) -> None: def bind_run_cache(self, cache: object) -> None: return None - def ensure_managed_database(self, name, *, schema, tables, keys=None, create_if_missing): - raise KeyError(name) + def ensure_managed_database(self, *, schema, tables, keys=None, create_if_missing): + raise KeyError("db_1") def close(self) -> None: return None diff --git a/tests/test_sql_client.py b/tests/test_sql_client.py index 6b7dc4b..e80d1dc 100644 --- a/tests/test_sql_client.py +++ b/tests/test_sql_client.py @@ -38,14 +38,14 @@ def __init__(self, table: pa.Table | None = None, tables: object = _MISSING) -> self._table = table if table is not None else CANNED # tables=None explicitly models a missing database (list raises). self._tables = ["spans"] if tables is _MISSING else tables - self.calls: list[tuple[str, str]] = [] + self.calls: list[str] = [] self.closed = False - def execute_sql(self, sql: str, *, database: str) -> pa.Table: - self.calls.append((sql, database)) + def execute_sql(self, sql: str) -> pa.Table: + self.calls.append(sql) return self._table - def list_managed_tables(self, database: str, *, schema: str): + def list_managed_tables(self, *, schema: str): if self._tables is None: raise HotdataTerminalError("database not found") return list(self._tables) @@ -120,8 +120,8 @@ def test_execute_query_yields_cursor() -> None: sc = _client(fake) with sc.execute_query('SELECT * FROM "default"."public"."spans"') as cur: assert cur.df().shape == (3, 3) - # SQL passed through verbatim, scoped by managed database_name. - assert fake.calls == [('SELECT * FROM "default"."public"."spans"', "db")] + # SQL passed through verbatim; the database scope is resolved by id internally. + assert fake.calls == ['SELECT * FROM "default"."public"."spans"'] def test_execute_sql_returns_rows() -> None: diff --git a/uv.lock b/uv.lock index 9829894..5cb0085 100644 --- a/uv.lock +++ b/uv.lock @@ -321,7 +321,7 @@ wheels = [ [[package]] name = "hotdata-dlt-destination" -version = "0.10.0" +version = "0.11.0" source = { editable = "." } dependencies = [ { name = "dlt" }, @@ -350,7 +350,7 @@ dev = [ requires-dist = [ { name = "dlt", specifier = ">=1.28.1,<1.29" }, { name = "hotdata", specifier = ">=0.8.0,<0.9" }, - { name = "hotdata-framework", specifier = ">=0.8.0,<0.9" }, + { name = "hotdata-framework", specifier = ">=0.9.0,<0.10" }, { name = "hotdata-ibis", marker = "extra == 'ibis'", specifier = ">=0.3.1,<0.4" }, { name = "ibis-framework", marker = "extra == 'ibis'", specifier = ">=12,<13" }, { name = "pandas", specifier = ">=2.0" }, @@ -369,16 +369,16 @@ dev = [ [[package]] name = "hotdata-framework" -version = "0.8.0" +version = "0.9.0" source = { registry = "https://pypi.org/simple" } dependencies = [ { name = "hotdata" }, { name = "pandas" }, { name = "pyarrow" }, ] -sdist = { url = "https://files.pythonhosted.org/packages/48/24/879cbc596520023d1757eef8fd73b482404fb98e98d48a19571e8932bec2/hotdata_framework-0.8.0.tar.gz", hash = "sha256:057ebe8818298f1e8b4ef4250a4f4266514c7b51f3a4eb79425a1548885348ce", size = 100984, upload-time = "2026-07-20T07:04:50.175Z" } +sdist = { url = "https://files.pythonhosted.org/packages/05/95/2e8970ba68bf0001d92d0044766c0e13cb09e7378aa003c637ad571786b0/hotdata_framework-0.9.0.tar.gz", hash = "sha256:a6497ef9fc41f104c89d5636e958a1feaf1c697d98d01436329afeb75f773583", size = 102244, upload-time = "2026-07-23T19:32:56.546Z" } wheels = [ - { url = "https://files.pythonhosted.org/packages/98/cd/c5078d826a2240e6d7c6ef5f558280772ee189991af15d0332fd11c5a67c/hotdata_framework-0.8.0-py3-none-any.whl", hash = "sha256:6ba899e9015b71945bf56df873c0486382d4f11763a57afc423338bd68abb30a", size = 16330, upload-time = "2026-07-20T07:04:49.089Z" }, + { url = "https://files.pythonhosted.org/packages/aa/52/d9bf9001d48262f84a712a3f07c07492964a4481d902117b8986b0dada9e/hotdata_framework-0.9.0-py3-none-any.whl", hash = "sha256:32e7a541823365b5d0e741c9209683406e3851de0434012631a7a7b16b029c1b", size = 16550, upload-time = "2026-07-23T19:32:55.064Z" }, ] [[package]]