feat: accept resolved ManagedDatabase in managed-table ops to skip read probe (#52)

Add HotdataClient._as_managed_database and widen list_managed_tables,
load_managed_table, add_managed_table, delete_managed_table,
delete_managed_database, and execute_sql to accept str | ManagedDatabase.
Passing an already-resolved ManagedDatabase (as returned by
create_managed_database) skips the get_database/list_databases read probe,
letting a create+load-scoped key bootstrap and load into a managed database
in a single run without a forbidden read. The name/id string path is
unchanged, and resolve_managed_database's 403->RuntimeError behavior is
preserved.

Additive change: minor version bump 0.8.0 -> 0.9.0.

Refs hotdata-dlt-destination#55, hotdata-dlt-destination#39.

Author: rohan-hotdata

CHANGELOG.md

@@ -8,6 +8,19 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ## [Unreleased]
 
 
+## [0.9.0] - 2026-07-23
+
+### Added
+
+- `list_managed_tables`, `load_managed_table`, `add_managed_table`,
+  `delete_managed_table`, `delete_managed_database`, and `execute_sql` accept an
+  already-resolved `ManagedDatabase` (as returned by `create_managed_database`)
+  in place of a name/id. When passed one, they skip the `get_database` /
+  `list_databases` read probe. This lets an API key scoped to create + load but
+  not read `/databases` bootstrap a managed database and load into it within a
+  single run: the caller holds the `ManagedDatabase` from `create` and drives
+  the load/add/query ops with zero reads. The name/id string path is unchanged.
+
 ## [0.8.0] - 2026-07-20
 
 ### Changed

CONTRACT.md

@@ -56,13 +56,14 @@ Adapters should import from `hotdata_framework` and treat this surface as the st
   adapters should pass `connection_id` when known.
 - `uploads()` returns the uploads API wrapper for parquet staging.
 - `list_managed_databases()` returns all databases via the `/databases` API.
-- `resolve_managed_database(name_or_id)` resolves a database by id (direct lookup) or description (list scan).
-- `create_managed_database(description=..., schema=..., tables=..., expires_at=...)` creates a database via the `/databases` API and optionally declares tables up front.
+- `resolve_managed_database(name_or_id)` resolves a database by id (direct lookup) or description (list scan). A `403` from `/databases` surfaces as `RuntimeError` (forbidden, not absent), preserving the underlying `ApiException` as `__cause__`.
+- `create_managed_database(description=..., schema=..., tables=..., expires_at=...)` creates a database via the `/databases` API and optionally declares tables up front. Returns a `ManagedDatabase` (id + `default_connection_id`) sufficient to load without a further read.
 - `delete_managed_database(name_or_id)` deletes a database via the `/databases` API.
 - `list_managed_tables(database, schema=...)` lists tables in a managed database.
 - `upload_parquet(path)` uploads a local parquet file and returns an upload id.
 - `load_managed_table(database, table, schema=..., upload_id=..., file=...)` publishes parquet data into a declared managed table.
 - `delete_managed_table(database, table, schema=...)` deletes a managed table.
+- The `database` argument of `list_managed_tables`, `load_managed_table`, `add_managed_table`, `delete_managed_table`, `delete_managed_database`, and `execute_sql` accepts a name/id **or** an already-resolved `ManagedDatabase`. Passing a `ManagedDatabase` skips the name/id read probe, so a create-scoped key that cannot read `/databases` can load into a database it just created.
 
 ### `QueryResult`
 

hotdata_framework/client.py

@@ -241,6 +241,19 @@ def resolve_managed_database(self, name_or_id: str) -> ManagedDatabase:
             raise RuntimeError(api_error_message(e)) from e
         return managed_database_from_detail(detail)
 
+    def _as_managed_database(self, database: str | ManagedDatabase) -> ManagedDatabase:
+        """Return ``database`` as-is if it is already a resolved ``ManagedDatabase``,
+        otherwise resolve it by name or id.
+
+        Passing an already-resolved ``ManagedDatabase`` (e.g. the value returned by
+        :meth:`create_managed_database`) skips the id/name read probe, so callers
+        whose API key may create but not read ``/databases`` can drive loads without
+        a forbidden read.
+        """
+        if isinstance(database, ManagedDatabase):
+            return database
+        return self.resolve_managed_database(database)
+
     def create_managed_database(
         self,
         description: str | None = None,
@@ -275,20 +288,20 @@ def create_managed_database(
             raise RuntimeError(api_error_message(e)) from e
         return managed_database_from_detail(created)
 
-    def delete_managed_database(self, name_or_id: str) -> None:
-        db = self.resolve_managed_database(name_or_id)
+    def delete_managed_database(self, name_or_id: str | ManagedDatabase) -> None:
+        db = self._as_managed_database(name_or_id)
         try:
             self._databases_api().delete_database(db.id)
         except ApiException as e:
             raise RuntimeError(api_error_message(e)) from e
 
     def list_managed_tables(
         self,
-        database: str,
+        database: str | ManagedDatabase,
         *,
         schema: str | None = None,
     ) -> list[ManagedTable]:
-        db = self.resolve_managed_database(database)
+        db = self._as_managed_database(database)
         rows: list[ManagedTable] = []
         for t in self.iter_tables(connection_id=db.default_connection_id):
             if schema is not None and t.var_schema != schema:
@@ -333,7 +346,7 @@ def upload_parquet(self, path: str) -> str:
 
     def load_managed_table(
         self,
-        database: str,
+        database: str | ManagedDatabase,
         table: str,
         *,
         schema: str = DEFAULT_SCHEMA,
@@ -344,7 +357,7 @@ def load_managed_table(
     ) -> LoadManagedTableResult:
         if (upload_id is None) == (file is None):
             raise ValueError("Exactly one of upload_id or file is required")
-        db = self.resolve_managed_database(database)
+        db = self._as_managed_database(database)
         if upload_id is not None:
             resolved_upload_id = upload_id
         else:
@@ -374,7 +387,7 @@ def load_managed_table(
 
     def add_managed_table(
         self,
-        database: str,
+        database: str | ManagedDatabase,
         table: str,
         *,
         schema: str = DEFAULT_SCHEMA,
@@ -387,7 +400,7 @@ def add_managed_table(
         schema after creation without recreating it. ``key`` sets the
         row-identity columns for delete/update/upsert; omit for keyless.
         """
-        db = self.resolve_managed_database(database)
+        db = self._as_managed_database(database)
         request = AddManagedTableRequest(name=table, key=list(key or []))
         try:
             self._databases_api().add_database_table(db.id, schema, request)
@@ -403,12 +416,12 @@ def add_managed_table(
 
     def delete_managed_table(
         self,
-        database: str,
+        database: str | ManagedDatabase,
         table: str,
         *,
         schema: str = DEFAULT_SCHEMA,
     ) -> None:
-        db = self.resolve_managed_database(database)
+        db = self._as_managed_database(database)
         try:
             self.connections().delete_managed_table(db.default_connection_id, schema, table)
         except ApiException as e:
@@ -569,16 +582,19 @@ def _wait_result_ready(
             f"(last status: {getattr(last, 'status', None)})"
         )
 
-    def execute_sql(self, sql: str, *, database: str | None = None) -> QueryResult:
+    def execute_sql(
+        self, sql: str, *, database: str | ManagedDatabase | None = None
+    ) -> QueryResult:
         """Execute SQL and return a :class:`QueryResult`.
 
-        Pass ``database`` to scope the query to a managed database.  The name
-        is resolved to a database ID once before the retry loop, and the
+        Pass ``database`` to scope the query to a managed database.  A name or
+        id is resolved to a database ID once before the retry loop; an
+        already-resolved ``ManagedDatabase`` is used as-is (no read probe).  The
         ``X-Database-Id`` header is sent with every attempt.  Inside a managed
         database the built-in catalog is always ``"default"``, so table
         references should use ``"default"."<schema>"."<table>"``.
         """
-        database_id = self.resolve_managed_database(database).id if database else None
+        database_id = self._as_managed_database(database).id if database else None
         last_err: BaseException | None = None
         for attempt in range(3):
             try:

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "hotdata-framework"
-version = "0.8.0"
+version = "0.9.0"
 description = "Python framework for building Hotdata integrations: workspace/session runtime, query execution, and managed databases"
 readme = "README.md"
 requires-python = ">=3.10"

tests/test_client.py

@@ -4,11 +4,134 @@
 from unittest.mock import patch
 
 import pytest
+from hotdata.exceptions import ForbiddenException
 
 from hotdata_framework.client import HotdataClient
+from hotdata_framework.databases import ManagedDatabase
 from hotdata_framework.env import normalize_host, pick_workspace, resolve_workspace_selection
 
 
+class _ForbiddenDatabasesApi:
+    """A `/databases` API that a create-scoped key would see: every read is 403,
+    while the declare-table write succeeds. Counts reads so tests can assert none
+    happened."""
+
+    def __init__(self) -> None:
+        self.read_calls = 0
+        self.add_calls: list[tuple[str, str, str]] = []
+
+    def get_database(self, database_id: str):
+        self.read_calls += 1
+        raise ForbiddenException(status=403)
+
+    def list_databases(self):
+        self.read_calls += 1
+        raise ForbiddenException(status=403)
+
+    def add_database_table(self, database_id, var_schema, request):
+        self.add_calls.append((database_id, var_schema, request.name))
+        return SimpleNamespace(
+            connection_id="conn", var_schema=var_schema, table=request.name
+        )
+
+
+class _FakeConnectionsApi:
+    def __init__(self) -> None:
+        self.load_calls: list[tuple[str, str, str]] = []
+
+    def load_managed_table(self, connection_id, schema, table, request):
+        self.load_calls.append((connection_id, schema, table))
+        return SimpleNamespace(
+            connection_id=connection_id,
+            schema_name=schema,
+            table_name=table,
+            row_count=3,
+        )
+
+
+def test_load_managed_table_with_object_skips_read_probe():
+    client = HotdataClient("k", "ws", host="https://api.hotdata.dev")
+    db = ManagedDatabase(id="db_1", description="mydb", default_connection_id="conn_1")
+    databases = _ForbiddenDatabasesApi()
+    connections = _FakeConnectionsApi()
+
+    with (
+        patch.object(client, "_databases_api", return_value=databases),
+        patch.object(client, "connections", return_value=connections),
+    ):
+        result = client.load_managed_table(db, "orders", schema="public", upload_id="up_1")
+
+    assert databases.read_calls == 0
+    assert connections.load_calls == [("conn_1", "public", "orders")]
+    assert result.full_name == "db_1.public.orders"
+    assert result.row_count == 3
+
+
+def test_add_managed_table_with_object_skips_read_probe():
+    client = HotdataClient("k", "ws", host="https://api.hotdata.dev")
+    db = ManagedDatabase(id="db_1", description="mydb", default_connection_id="conn_1")
+    databases = _ForbiddenDatabasesApi()
+
+    with patch.object(client, "_databases_api", return_value=databases):
+        result = client.add_managed_table(db, "orders", schema="public")
+
+    assert databases.read_calls == 0
+    assert databases.add_calls == [("db_1", "public", "orders")]
+    assert result.full_name == "db_1.public.orders"
+
+
+def test_execute_sql_with_object_skips_read_probe():
+    from hotdata.models.query_response import QueryResponse as _QR
+
+    client = HotdataClient("k", "ws", host="https://api.hotdata.dev")
+    db = ManagedDatabase(id="db_abc", description="mydb", default_connection_id="conn_1")
+    databases = _ForbiddenDatabasesApi()
+
+    class FakeQueryApi:
+        def __init__(self):
+            self.calls: list[dict] = []
+
+        def query(self, request, **kwargs):
+            self.calls.append(kwargs)
+            return _QR(
+                columns=["n"],
+                rows=[[1]],
+                row_count=1,
+                preview_row_count=1,
+                truncated=False,
+                nullable=[False],
+                result_id="res_1",
+                query_run_id="qrun_1",
+                execution_time_ms=1,
+            )
+
+    fake_q = FakeQueryApi()
+    with (
+        patch.object(client, "_query_api", return_value=fake_q),
+        patch.object(client, "_databases_api", return_value=databases),
+    ):
+        client.execute_sql("SELECT 1", database=db)
+
+    assert databases.read_calls == 0
+    assert fake_q.calls == [{"x_database_id": "db_abc"}]
+
+
+def test_load_managed_table_with_name_still_resolves():
+    client = HotdataClient("k", "ws", host="https://api.hotdata.dev")
+    connections = _FakeConnectionsApi()
+    resolved = ManagedDatabase(id="db_1", description="mydb", default_connection_id="conn_1")
+
+    with (
+        patch.object(client, "resolve_managed_database", return_value=resolved) as resolve,
+        patch.object(client, "connections", return_value=connections),
+    ):
+        result = client.load_managed_table("mydb", "orders", schema="public", upload_id="up_1")
+
+    resolve.assert_called_once_with("mydb")
+    assert connections.load_calls == [("conn_1", "public", "orders")]
+    assert result.full_name == "db_1.public.orders"
+
+
 def _clear_workspace_env(monkeypatch: pytest.MonkeyPatch) -> None:
     monkeypatch.delenv("HOTDATA_WORKSPACE", raising=False)