feat: expose native PyIceberg catalog access

INSTALL-AND-REFERENCE.md

@@ -128,7 +128,54 @@ pip install "tower[ai]"
 pip install "tower[iceberg]"
 ```
 
-- `tower.tables(...)`: load, create, update, and delete Iceberg table data
+- `tower.load_catalog(...)`: use a Tower-managed catalog through native PyIceberg APIs
+- `tower.tables(...)`: load, create, update, and delete Iceberg table data through Tower's convenience API
+
+#### Native PyIceberg access
+
+Set `TOWER_JWT` or `TOWER_API_KEY` in the Python process, then load a catalog for
+the target environment:
+
+```python
+import tower
+
+catalog = tower.load_catalog(
+    "analytics",
+    environment="production",
+)
+
+catalog.list_namespaces()
+events = catalog.load_table(("analytics", "events"))
+```
+
+`tower.load_catalog()` returns an unwrapped `pyiceberg.catalog.Catalog`. Tower
+first looks for the named catalog in the target environment, then uses the
+same-named catalog from the shared `default` environment when it is not locally
+defined. Only Tower-managed catalogs are supported by this API.
+
+Access is read-only by default. Request write credentials explicitly for
+namespace or table mutations:
+
+```python
+catalog = tower.load_catalog(
+    "analytics",
+    environment="production",
+    mode="read-write",
+)
+```
+
+When both authentication variables are set, `TOWER_JWT` takes precedence over
+`TOWER_API_KEY`. `tower login` authenticates the CLI and MCP server only; it does
+not authenticate Python SDK calls.
+
+The returned catalog has one fixed temporary provider token and does not renew
+itself. If the token expires, the original PyIceberg or provider error is
+preserved. Call `tower.load_catalog()` again to obtain a fresh handle. Missing,
+forbidden, unsupported, and connection failures are available as distinct
+exceptions in `tower.exceptions`. A missing `default` catalog is not created
+automatically.
+
+#### Tower table convenience API
 
 Delete filters are SQL-like strings or native PyIceberg boolean expressions. The
 `Table.column()` builder creates composable PyIceberg predicates:

README.md

@@ -122,7 +122,7 @@ If you find Tower useful, consider giving the repo a ⭐.
 - **Consistent execution environment everywhere** - `tower run` on Tower serverless or your own compute.
 - **Deploy in under 30 seconds** - `tower deploy` packages and ships.
 - **Secrets** - CLI-managed; injected as env vars in runner only (E2E encrypted).
-- **Optional AI inference, Iceberg or dbt** - `tower[ai]`, `tower[iceberg]` or `tower[dbt]`; [details](INSTALL-AND-REFERENCE.md#optional-features).
+- **Optional AI inference, native PyIceberg catalogs, or dbt** - `tower[ai]`, `tower[iceberg]` or `tower[dbt]`; [details](INSTALL-AND-REFERENCE.md#optional-features).
 - **MCP server** - Deploy and launch runs from AI coding assistants; [details](https://docs.tower.dev/docs/reference/mcp-server).
 
 ---

src/tower/__init__.py

@@ -30,6 +30,7 @@
 from ._features import override_get_attr, get_available_features, is_feature_enabled
 
 if TYPE_CHECKING:
+    from ._storage import load_catalog as load_catalog
     from ._tables import tables as tables
     from ._llms import llms as llms
     from ._dbt import dbt as dbt

src/tower/_features.py

@@ -42,7 +42,7 @@ def get_package_version(pkg_name: str) -> Optional[str]:
 # export_type can be "function" (return specific function) or "module" (return entire module)
 _feature_modules: Dict[str, tuple[str, List[str], str]] = {
     "ai": ("_llms", ["llms"], "function"),
-    "iceberg": ("_tables", ["tables"], "function"),
+    "iceberg": ("_iceberg", ["load_catalog", "tables"], "function"),
     "dbt": ("_dbt", ["dbt"], "function"),
 }
 

src/tower/_iceberg.py

@@ -0,0 +1,6 @@
+"""Public exports for Tower's optional Apache Iceberg feature."""
+
+from ._storage import load_catalog
+from ._tables import tables
+
+__all__ = ["load_catalog", "tables"]

src/tower/_storage.py

@@ -7,25 +7,24 @@
 from dataclasses import dataclass, field
 from datetime import datetime, timedelta, timezone
 from threading import Lock
-from typing import TYPE_CHECKING
+from typing import Literal
 
 import httpx
-
-if TYPE_CHECKING:
-    from pyiceberg.catalog import Catalog
+from pyiceberg.catalog import Catalog
 
 from ._context import TowerContext
 from .exceptions import (
+    StorageAuthenticationError,
+    StorageCatalogNotFoundError,
     StorageConnectionError,
     StorageError,
     StorageInvalidCredentialError,
     StorageMissingAuthenticationError,
+    StoragePermissionError,
+    StorageUnsupportedCatalogError,
 )
 from .tower_api_client import AuthenticatedClient
 from .tower_api_client.api.default import describe_catalog as describe_catalog_api
-from .tower_api_client.api.default import (
-    describe_default_catalog as describe_default_catalog_api,
-)
 from .tower_api_client.api.default import (
     vend_catalog_credentials as vend_catalog_credentials_api,
 )
@@ -46,7 +45,6 @@
 # retry only after this period
 CATALOG_TYPE_FAILURE_CACHE_TTL_SECONDS = 30.0
 DEFAULT_STORAGE_TIMEOUT_SECONDS = 30.0
-DEFAULT_CATALOG_PROVISION_RETRY_DELAYS = (0.25, 0.5, 1.0, 2.0)
 DEFAULT_CATALOG_NAME = "default"
 DEFAULT_ENVIRONMENT_NAME = "default"
 TOWER_CATALOG_TYPE = "tower-catalog"
@@ -208,10 +206,10 @@ def _resolve_catalog_access(
             return flight.result()
 
         try:
-            response = _vend_with_default_catalog_fallback(
-                self,
+            response = _unwrap_vend_result(
+                self._request_catalog_credentials(name, mode),
                 name,
-                mode,
+                target_environment,
             )
             credentials = response.credentials
             if credentials.mode != mode:
@@ -301,16 +299,58 @@ class _CachedCatalogType:
 _catalog_type_cache: dict[_CatalogTypeCacheKey, _CachedCatalogType] = {}
 
 
-def get_tower_catalog(
+def load_catalog(
     name: str = DEFAULT_CATALOG_NAME,
+    *,
     environment: str | None = None,
-    mode: str = "read",
+    mode: Literal["read", "read-write"] = "read",
 ) -> Catalog:
+    """Load a Tower-managed catalog as a native PyIceberg catalog.
+
+    The SDK authenticates from the process environment. ``TOWER_JWT`` takes
+    precedence over ``TOWER_API_KEY`` when both are set. A Tower CLI login does
+    not authenticate this Python API.
+
+    The returned catalog uses one fixed set of temporary provider credentials.
+    It does not retain a Tower resolver or renew itself. If those credentials
+    expire, the provider's original error is raised; call ``load_catalog`` again
+    to obtain a fresh catalog handle.
+
+    Args:
+        name: Name of the Tower-managed catalog. Defaults to ``"default"``.
+        environment: Target Tower environment. Tower first resolves the catalog
+            in this environment, then in the shared ``default`` environment. If
+            omitted, the configured Tower environment is used.
+        mode: ``"read"`` for read-only credentials, or ``"read-write"`` for
+            namespace and table mutations. Defaults to ``"read"``.
+
+    Returns:
+        A caller-owned :class:`pyiceberg.catalog.Catalog` configured with the
+        resolved REST endpoint, warehouse, and temporary provider token.
+
+    Raises:
+        StorageMissingAuthenticationError: If neither supported Tower
+            credential is configured.
+        StorageAuthenticationError: If Tower rejects the configured credential.
+        StorageCatalogNotFoundError: If the environment or catalog is not found.
+        StoragePermissionError: If the credential lacks the requested access.
+        StorageUnsupportedCatalogError: If the catalog is not Tower-managed or
+            otherwise cannot vend native PyIceberg access.
+        StorageConnectionError: If Tower's control-plane API cannot be reached.
+        TypeError: If ``environment`` is not a string or ``None``.
+        ValueError: If ``environment`` is blank or ``mode`` is invalid.
+
+    Examples:
+        >>> import tower
+        >>> catalog = tower.load_catalog(
+        ...     "analytics",
+        ...     environment="production",
+        ... )
+        >>> catalog.list_namespaces()
     """
-    Load a PyIceberg REST catalog using short-lived credentials vended by Tower.
-    """
-    credentials = get_tower_catalog_credentials(name, environment, mode)
-    return load_vended_catalog(name, credentials)
+    storage_resolver = _StorageResolver(environment=environment)
+    access = storage_resolver._resolve_catalog_access(name, mode)
+    return load_vended_catalog(name, access.to_credentials())
 
 
 def get_tower_catalog_credentials(
@@ -335,32 +375,6 @@ def load_vended_catalog(name: str, credentials: CatalogCredentials) -> Catalog:
     )
 
 
-def _vend_with_default_catalog_fallback(
-    storage_resolver: _StorageResolver,
-    name: str,
-    mode: str,
-) -> VendCatalogCredentialsResponse:
-    environment = storage_resolver._target_environment
-    result = storage_resolver._request_catalog_credentials(name, mode)
-    if not _is_not_found(result):
-        return _unwrap_vend_result(result, name, environment)
-
-    if name == DEFAULT_CATALOG_NAME and environment == DEFAULT_ENVIRONMENT_NAME:
-        _ensure_legacy_default_catalog(storage_resolver)
-        for delay in DEFAULT_CATALOG_PROVISION_RETRY_DELAYS:
-            time.sleep(delay)
-            result = storage_resolver._request_catalog_credentials(name, mode)
-            if not _is_not_found(result):
-                return _unwrap_vend_result(result, name, environment)
-            _ensure_legacy_default_catalog(storage_resolver)
-
-        return _unwrap_vend_result(result, name, environment)
-
-    raise RuntimeError(
-        f"Tower catalog {name} does not exist in environment {environment}."
-    )
-
-
 def _describe_tower_catalog_type(
     ctx: TowerContext, name: str, environment: str
 ) -> str | None:
@@ -437,15 +451,6 @@ def _failed_catalog_type_cache_entry() -> _CachedCatalogType:
     )
 
 
-def _ensure_legacy_default_catalog(storage_resolver: _StorageResolver) -> None:
-    try:
-        with storage_resolver._new_client() as client:
-            describe_default_catalog_api.sync(client=client)
-    except Exception:
-        # The following vend retry will surface the actionable backend/auth error.
-        return
-
-
 def _unwrap_vend_result(
     result: ErrorModel | VendCatalogCredentialsResponse | None,
     name: str,
@@ -456,14 +461,25 @@ def _unwrap_vend_result(
 
     if isinstance(result, ErrorModel):
         detail = _error_text(result)
-        raise RuntimeError(
-            f"Failed to vend credentials for Tower catalog {name} "
-            f"in environment {environment}: {detail}"
+        message = (
+            f"Could not load Tower catalog {name!r} "
+            f"in environment {environment!r}: {detail}"
         )
-
-    raise RuntimeError(
-        f"Failed to vend credentials for Tower catalog {name} "
-        f"in environment {environment}."
+        error_type: type[StorageError]
+        if result.status == 401:
+            error_type = StorageAuthenticationError
+        elif result.status == 403:
+            error_type = StoragePermissionError
+        elif result.status == 404:
+            error_type = StorageCatalogNotFoundError
+        elif result.status == 422:
+            error_type = StorageUnsupportedCatalogError
+        else:
+            error_type = StorageError
+        raise error_type(message)
+
+    raise StorageError(
+        f"Could not load Tower catalog {name!r} in environment {environment!r}."
     )
 
 
@@ -481,10 +497,6 @@ def _vend_mode(mode: str) -> VendCatalogCredentialsBodyMode:
     )
 
 
-def _is_not_found(result: ErrorModel | VendCatalogCredentialsResponse | None) -> bool:
-    return isinstance(result, ErrorModel) and result.status == 404
-
-
 def _error_text(error: ErrorModel) -> str:
     for value in (error.detail, error.title):
         if not isinstance(value, Unset) and value:

src/tower/exceptions.py

@@ -63,3 +63,15 @@ class StorageInvalidCredentialError(StorageAuthenticationError):
 
 class StorageConnectionError(StorageError):
     """Tower's control-plane API could not be reached."""
+
+
+class StorageCatalogNotFoundError(StorageError):
+    """The target environment or requested catalog could not be found."""
+
+
+class StoragePermissionError(StorageError):
+    """The Tower credential lacks permission for the requested catalog access."""
+
+
+class StorageUnsupportedCatalogError(StorageError):
+    """Tower cannot vend native PyIceberg access for the requested catalog."""