feat!: address managed databases by id only, never by name (#39, #55)

.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

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 <id> … set database_id=<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

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)) |

docs/runbook.md

@@ -31,7 +31,10 @@
   uv run hotdata-dlt-demo --workspace-id <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 <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
 

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.<table>
         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

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",
 ]

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,

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,

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 '<unset — will create by name>'}")
     print(f"database_name={config.database_name}")
     print(f"schema={config.schema}")
     print(f"write_disposition={config.write_disposition}")