diff --git a/.github/workflows/build-binaries.yml b/.github/workflows/build-binaries.yml index fe7962ae..bf9e6eb5 100644 --- a/.github/workflows/build-binaries.yml +++ b/.github/workflows/build-binaries.yml @@ -197,8 +197,8 @@ jobs: runs-on: ubuntu-latest strategy: matrix: - target: - - x86_64-unknown-linux-gnu + platform: + - target: x86_64-unknown-linux-gnu python-version: ["3.12", "3.13", "3.14"] steps: - uses: actions/checkout@v6 @@ -212,7 +212,7 @@ jobs: - name: Build wheel uses: PyO3/maturin-action@v1.50.1 with: - target: ${{ matrix.target }} + target: ${{ matrix.platform.target }} manylinux: auto args: --release --locked --out dist -i python${{ matrix.python-version }} before-script-linux: | @@ -225,7 +225,7 @@ jobs: sudo apt-get install -y libssl-dev openssl pkg-config fi - name: Test wheel - if: ${{ startsWith(matrix.target, 'x86_64') }} + if: ${{ startsWith(matrix.platform.target, 'x86_64') }} run: | PYTAG="cp$(python -c "import sys; print(f'{sys.version_info.major}{sys.version_info.minor}')")" pip install dist/*-${PYTAG}-*.whl --force-reinstall @@ -234,7 +234,7 @@ jobs: - name: Upload wheel uses: actions/upload-artifact@v6 with: - name: wheels-${{ matrix.target }} + name: wheels-${{ matrix.platform.target }}-py${{ matrix.python-version }} path: dist linux-cross: @@ -302,8 +302,8 @@ jobs: runs-on: ubuntu-latest strategy: matrix: - target: - - x86_64-unknown-linux-musl + platform: + - target: x86_64-unknown-linux-musl python-version: ["3.12", "3.13", "3.14"] steps: - uses: actions/checkout@v6 @@ -317,11 +317,11 @@ jobs: - name: Build wheel uses: PyO3/maturin-action@v1.50.1 with: - target: ${{ matrix.target }} + target: ${{ matrix.platform.target }} manylinux: musllinux_1_2 args: --release --locked --out dist -i python${{ matrix.python-version }} - name: Test wheel - if: matrix.target == 'x86_64-unknown-linux-musl' + if: matrix.platform.target == 'x86_64-unknown-linux-musl' uses: addnab/docker-run-action@v3 with: image: python:${{ matrix.python-version}}-alpine @@ -334,7 +334,7 @@ jobs: - name: Upload wheel uses: actions/upload-artifact@v6 with: - name: wheels-${{ matrix.target }} + name: wheels-${{ matrix.platform.target }}-py${{ matrix.python-version }} path: dist musllinux-cross: @@ -376,5 +376,5 @@ jobs: - name: Upload wheel uses: actions/upload-artifact@v6 with: - name: wheels-${{ matrix.platform.target }} + name: wheels-${{ matrix.platform.target }}-py${{ matrix.python-version }} path: dist diff --git a/.github/workflows/publish-npm.yml b/.github/workflows/publish-npm.yml index 385434dc..9c1f3ff5 100644 --- a/.github/workflows/publish-npm.yml +++ b/.github/workflows/publish-npm.yml @@ -31,7 +31,6 @@ jobs: - uses: actions/setup-node@v4 with: node-version: 24 - registry-url: 'https://registry.npmjs.org' - name: Upgrade npm to latest patch run: npm install -g npm@latest - name: Publish to npm diff --git a/Cargo.lock b/Cargo.lock index c7dc82d8..013fb4b7 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -491,7 +491,7 @@ dependencies = [ [[package]] name = "config" -version = "0.3.68" +version = "0.3.69" dependencies = [ "base64", "chrono", @@ -634,7 +634,7 @@ dependencies = [ [[package]] name = "crypto" -version = "0.3.68" +version = "0.3.69" dependencies = [ "aes-gcm", "base64", @@ -3408,7 +3408,7 @@ dependencies = [ [[package]] name = "testutils" -version = "0.3.68" +version = "0.3.69" dependencies = [ "pem", "rsa", @@ -3663,7 +3663,7 @@ checksum = "5d99f8c9a7727884afe522e9bd5edbfc91a3312b36a77b5fb8926e4c31a41801" [[package]] name = "tower" -version = "0.3.68" +version = "0.3.69" dependencies = [ "config", "pyo3", @@ -3691,7 +3691,7 @@ dependencies = [ [[package]] name = "tower-api" -version = "0.3.68" +version = "0.3.69" dependencies = [ "reqwest", "serde", @@ -3703,7 +3703,7 @@ dependencies = [ [[package]] name = "tower-cmd" -version = "0.3.68" +version = "0.3.69" dependencies = [ "axum", "bytes", @@ -3775,7 +3775,7 @@ checksum = "121c2a6cda46980bb0fcd1647ffaf6cd3fc79a013de288782836f6df9c48780e" [[package]] name = "tower-package" -version = "0.3.68" +version = "0.3.69" dependencies = [ "async-compression", "flate2", @@ -3800,7 +3800,7 @@ dependencies = [ [[package]] name = "tower-runtime" -version = "0.3.68" +version = "0.3.69" dependencies = [ "async-trait", "chrono", @@ -3824,7 +3824,7 @@ checksum = "8df9b6e13f2d32c91b9bd719c00d1958837bc7dec474d94952798cc8e69eeec3" [[package]] name = "tower-telemetry" -version = "0.3.68" +version = "0.3.69" dependencies = [ "tracing", "tracing-appender", @@ -3833,7 +3833,7 @@ dependencies = [ [[package]] name = "tower-uv" -version = "0.3.68" +version = "0.3.69" dependencies = [ "async-compression", "async_zip", @@ -3853,7 +3853,7 @@ dependencies = [ [[package]] name = "tower-version" -version = "0.3.68" +version = "0.3.69" dependencies = [ "anyhow", "chrono", diff --git a/Cargo.toml b/Cargo.toml index 138b9ca0..f89fb323 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -4,7 +4,7 @@ resolver = "2" [workspace.package] edition = "2021" -version = "0.3.68" +version = "0.3.69" description = "Tower is the best way to host Python data apps in production" rust-version = "1.81" authors = ["Brad Heller ", "Ben Lovell "] diff --git a/crates/tower-package/Cargo.toml b/crates/tower-package/Cargo.toml index ce0ddb3f..a7b14019 100644 --- a/crates/tower-package/Cargo.toml +++ b/crates/tower-package/Cargo.toml @@ -5,6 +5,7 @@ authors = { workspace = true } edition = { workspace = true } rust-version = { workspace = true } license = { workspace = true } +repository = { workspace = true } [lib] crate-type = ["cdylib", "rlib"] diff --git a/crates/tower-uv/src/lib.rs b/crates/tower-uv/src/lib.rs index de35cedc..7af6006c 100644 --- a/crates/tower-uv/src/lib.rs +++ b/crates/tower-uv/src/lib.rs @@ -302,6 +302,16 @@ impl Uv { } } + // Without an explicit cache dir, uv's default cache may be on a different + // filesystem than the target venv, so hardlinking fails and uv warns before + // falling back to a copy. Force copy mode to skip the failed attempt and the + // warning. + fn apply_link_mode(&self, cmd: &mut Command) { + if self.cache_dir.is_none() { + cmd.env("UV_LINK_MODE", "copy"); + } + } + pub async fn venv( &self, cwd: &PathBuf, @@ -361,6 +371,7 @@ impl Uv { if let Some(dir) = &self.cache_dir { cmd.arg("--cache-dir").arg(dir); } + self.apply_link_mode(&mut cmd); let child = cmd.spawn()?; @@ -406,6 +417,7 @@ impl Uv { .arg("never") .arg("pip") .arg("install"); + self.apply_link_mode(&mut cmd); cmd } @@ -494,6 +506,8 @@ impl Uv { // Need to do this after env_clear intentionally. cmd.envs(env_vars); + // Also after env_clear so protected mode doesn't wipe it. + self.apply_link_mode(&mut cmd); if let Some(dir) = &self.cache_dir { cmd.arg("--cache-dir").arg(dir); diff --git a/pyproject.toml b/pyproject.toml index 24768dd8..829727a3 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "maturin" [project] name = "tower" -version = "0.3.68" +version = "0.3.69" description = "Tower CLI and runtime environment for Tower." authors = [{ name = "Tower Computing Inc.", email = "brad@tower.dev" }] readme = "README.md" diff --git a/src/tower/_storage.py b/src/tower/_storage.py index 01477b04..bff78bc1 100644 --- a/src/tower/_storage.py +++ b/src/tower/_storage.py @@ -1,6 +1,7 @@ from __future__ import annotations import hashlib +import logging import time from dataclasses import dataclass from datetime import datetime, timedelta, timezone @@ -9,6 +10,7 @@ from ._client import _env_client from ._context import TowerContext +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, ) @@ -17,6 +19,7 @@ ) from .tower_api_client.models import ( CatalogCredentials, + DescribeCatalogResponse, ErrorModel, VendCatalogCredentialsBody, VendCatalogCredentialsBodyMode, @@ -25,9 +28,16 @@ from .tower_api_client.types import UNSET, Unset CREDENTIAL_REFRESH_WINDOW = timedelta(minutes=5) +# how long to wait for a catalog type describe request to complete +CATALOG_TYPE_DESCRIBE_TIMEOUT_SECONDS = 2.0 +# only cache failed catalog type describe requests for this long +# retry only after this period +CATALOG_TYPE_FAILURE_CACHE_TTL_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" +logger = logging.getLogger("tower.storage") @dataclass @@ -39,7 +49,14 @@ def is_usable(self, now: datetime) -> bool: return now < expires_at - CREDENTIAL_REFRESH_WINDOW +@dataclass +class _CachedCatalogType: + catalog_type: str | None + retry_at: float | None = None + + _credential_cache: dict[tuple[str, str, str, str, str], _CachedCredentials] = {} +_catalog_type_cache: dict[tuple[str, str, str], _CachedCatalogType] = {} def get_tower_catalog( @@ -123,6 +140,65 @@ def _vend_catalog_credentials( ) +def _describe_tower_catalog_type( + ctx: TowerContext, name: str, environment: str +) -> str | None: + if not (ctx.api_key or ctx.jwt): + return None + + cache_key = (ctx.tower_url, name, environment) + cached = _catalog_type_cache.get(cache_key) + if cached is not None: + if cached.retry_at is None: + return cached.catalog_type + + if time.monotonic() < cached.retry_at: + return None + + _catalog_type_cache.pop(cache_key, None) + + try: + result = describe_catalog_api.sync( + name=name, + client=_env_client(ctx, timeout=CATALOG_TYPE_DESCRIBE_TIMEOUT_SECONDS), + environment=environment, + ) + except Exception: + logger.debug( + "Failed to describe Tower catalog %r in environment %r; " + "falling back to PyIceberg catalog configuration detection.", + name, + environment, + exc_info=True, + ) + _catalog_type_cache[cache_key] = _failed_catalog_type_cache_entry() + return None + + if isinstance(result, DescribeCatalogResponse): + catalog_type = result.catalog.type_ + _catalog_type_cache[cache_key] = _CachedCatalogType(catalog_type=catalog_type) + return catalog_type + + if isinstance(result, ErrorModel): + logger.debug( + "Tower catalog describe for %r in environment %r returned %s; " + "falling back to PyIceberg catalog configuration detection.", + name, + environment, + _error_text(result), + ) + + _catalog_type_cache[cache_key] = _failed_catalog_type_cache_entry() + return None + + +def _failed_catalog_type_cache_entry() -> _CachedCatalogType: + return _CachedCatalogType( + catalog_type=None, + retry_at=time.monotonic() + CATALOG_TYPE_FAILURE_CACHE_TTL_SECONDS, + ) + + def _ensure_legacy_default_catalog(ctx: TowerContext) -> None: try: response = describe_default_catalog_api.sync_detailed(client=_env_client(ctx)) @@ -210,3 +286,4 @@ def _ensure_aware(value: datetime) -> datetime: def _clear_credential_cache() -> None: _credential_cache.clear() + _catalog_type_cache.clear() diff --git a/src/tower/_tables.py b/src/tower/_tables.py index 0f56540b..2b0d8401 100644 --- a/src/tower/_tables.py +++ b/src/tower/_tables.py @@ -1,5 +1,6 @@ from __future__ import annotations +import os from dataclasses import dataclass from typing import List, Optional, TypeVar, Union @@ -20,7 +21,12 @@ from pyiceberg.table import Table as IcebergTable from ._context import TowerContext -from ._storage import get_tower_catalog_credentials, load_vended_catalog +from ._storage import ( + TOWER_CATALOG_TYPE, + _describe_tower_catalog_type, + get_tower_catalog_credentials, + load_vended_catalog, +) from .tower_api_client.models import CatalogCredentials from .utils.pyarrow import ( convert_pyarrow_expressions, @@ -60,6 +66,45 @@ def _load_tower_catalog( return load_vended_catalog(name, credentials), _vended_catalog_identity(credentials) +def _pyiceberg_catalog_env_prefix(name: str) -> str: + catalog_name = name.replace("-", "_").replace(".", "_").replace(":", "_").upper() + return f"PYICEBERG_CATALOG__{catalog_name}__" + + +def _has_pyiceberg_catalog_config(name: str) -> bool: + try: + from pyiceberg.catalog import _ENV_CONFIG + + if _ENV_CONFIG.get_catalog_config(name) is not None: + return True + except Exception: + pass + + prefix = _pyiceberg_catalog_env_prefix(name) + return any(key.upper().startswith(prefix) for key in os.environ) + + +def _should_vend_tower_credentials( + ctx: TowerContext, + name: str, + environment: str, + tower_credentials: Optional[bool], +) -> bool: + """Choose Tower vending only for managed catalogs unless explicitly overridden. + + BYO catalogs such as S3 Tables already receive PyIceberg config from the runner, + so the default path must preserve that instead of forcing Tower vending. + """ + if tower_credentials is not None: + return tower_credentials + + catalog_type = _describe_tower_catalog_type(ctx, name, environment) + if catalog_type is not None: + return catalog_type == TOWER_CATALOG_TYPE + + return not _has_pyiceberg_catalog_config(name) + + class Table: """ `Table` is a wrapper around an Iceberg table. It provides methods to read and @@ -743,10 +788,11 @@ def tables( namespace (Optional[str], optional): The namespace in which the table exists or should be created. If not provided, a default namespace will be used. tower_credentials (Optional[bool], optional): Credential resolution for string - catalogs. By default (None) and when True, credentials are vended from - Tower. Set False to fall back to existing PyIceberg configuration (the - legacy ``PYICEBERG_CATALOG__*`` env vars) — a temporary rollback hatch. - Ignored when a Catalog instance is passed. + catalogs. By default (None), Tower-managed catalogs vend credentials and + other configured catalogs use existing PyIceberg configuration (including + runner-injected ``PYICEBERG_CATALOG__*`` env vars for S3 Tables). Set + True to force Tower credential vending or False to force PyIceberg + configuration. Ignored when a Catalog instance is passed. Returns: TableReference: A reference object that can be used to: @@ -788,19 +834,20 @@ def tables( vended_catalog_identity = None if isinstance(catalog, str): - # Tower-managed catalogs always resolve through credential vending. - # `tower_credentials=False` is a rollback hatch to the legacy - # PYICEBERG_CATALOG__* env-var config (still injected by the runner); - # remove it once that injection is gone. - if tower_credentials is False: - catalog = load_catalog(catalog) - else: + if _should_vend_tower_credentials( + ctx, + catalog, + ctx.environment, + tower_credentials, + ): catalog, vended_catalog_identity = _load_tower_catalog( catalog, environment=ctx.environment, mode="read", ) tower_vended = True + else: + catalog = load_catalog(catalog) return TableReference( ctx, diff --git a/tests/tower/test_storage.py b/tests/tower/test_storage.py index 2187b57f..43c70842 100644 --- a/tests/tower/test_storage.py +++ b/tests/tower/test_storage.py @@ -3,7 +3,9 @@ from tower._context import TowerContext from tower import _storage from tower.tower_api_client.models import ( + Catalog, CatalogCredentials, + DescribeCatalogResponse, ErrorModel, VendCatalogCredentialsResponse, ) @@ -181,3 +183,69 @@ def legacy_default(ctx): assert result is credentials assert len(legacy_calls) == 2 + + +def test_describe_tower_catalog_type_uses_timeout_and_recovers_after_cooldown( + monkeypatch, +): + _storage._clear_credential_cache() + ctx = TowerContext( + tower_url="https://api.example.com", + environment="production", + api_key="api-key", + ) + now = {"value": 100.0} + calls = [] + + def describe_catalog_api_sync(name, client, environment): + calls.append((name, environment, client._timeout)) + if len(calls) == 1: + raise TimeoutError("describe timed out") + + return DescribeCatalogResponse( + catalog=Catalog( + created_at=datetime.now(timezone.utc), + environment=environment, + name=name, + properties=[], + type_="s3-tables", + ) + ) + + monkeypatch.setattr(_storage.time, "monotonic", lambda: now["value"]) + monkeypatch.setattr( + _storage.describe_catalog_api, "sync", describe_catalog_api_sync + ) + + assert _storage._describe_tower_catalog_type(ctx, "s3-tables", "production") is None + assert _storage._describe_tower_catalog_type(ctx, "s3-tables", "production") is None + assert calls == [ + ( + "s3-tables", + "production", + _storage.CATALOG_TYPE_DESCRIBE_TIMEOUT_SECONDS, + ) + ] + + now["value"] += _storage.CATALOG_TYPE_FAILURE_CACHE_TTL_SECONDS + 0.1 + + assert ( + _storage._describe_tower_catalog_type(ctx, "s3-tables", "production") + == "s3-tables" + ) + assert ( + _storage._describe_tower_catalog_type(ctx, "s3-tables", "production") + == "s3-tables" + ) + assert calls == [ + ( + "s3-tables", + "production", + _storage.CATALOG_TYPE_DESCRIBE_TIMEOUT_SECONDS, + ), + ( + "s3-tables", + "production", + _storage.CATALOG_TYPE_DESCRIBE_TIMEOUT_SECONDS, + ), + ] diff --git a/tests/tower/test_tables.py b/tests/tower/test_tables.py index 38429a7d..946e1b11 100644 --- a/tests/tower/test_tables.py +++ b/tests/tower/test_tables.py @@ -19,8 +19,13 @@ # Imports the library under test import tower import tower._tables as tables_module +from tower import _storage from tower._context import TowerContext -from tower.tower_api_client.models import CatalogCredentials +from tower.tower_api_client.models import ( + Catalog, + CatalogCredentials, + DescribeCatalogResponse, +) class FakeLoadedTable: @@ -62,6 +67,18 @@ def patch_tower_context( return ctx +def make_describe_catalog_response(name: str, catalog_type: str): + return DescribeCatalogResponse( + catalog=Catalog( + created_at=datetime.datetime.now(datetime.timezone.utc), + environment="production", + name=name, + properties=[], + type_=catalog_type, + ) + ) + + def make_catalog_credentials(mode: str, token: str | None = None): return CatalogCredentials( catalog_uri="https://catalog.example.com", @@ -123,16 +140,20 @@ def sql_catalog(): @pytest.mark.parametrize( - ("tower_credentials", "expected_source"), + ("tower_credentials", "catalog_type", "has_pyiceberg_config", "expected_source"), [ - (None, "vend"), - (True, "vend"), - (False, "load_catalog"), + (None, "tower-catalog", True, "vend"), + (None, "s3-tables", True, "load_catalog"), + (None, None, True, "load_catalog"), + (None, None, False, "vend"), + (True, "s3-tables", True, "vend"), + (False, "tower-catalog", False, "load_catalog"), ], ) -def test_string_catalog_precedence(monkeypatch, tower_credentials, expected_source): - # Tower-managed catalogs vend by default and when forced; `tower_credentials=False` - # is the only path that falls back to existing PyIceberg configuration. +def test_string_catalog_precedence( + monkeypatch, tower_credentials, catalog_type, has_pyiceberg_config, expected_source +): + _storage._clear_credential_cache() patch_tower_context(monkeypatch) vend_catalog = FakeCatalog("vend") configured_catalog = FakeCatalog("configured") @@ -150,11 +171,27 @@ def load_catalog(name): calls.append(("load_catalog", name)) return configured_catalog + def describe_catalog_api_sync(name, client, environment): + calls.append(("describe_catalog", name, environment)) + if catalog_type is None: + return None + return make_describe_catalog_response(name, catalog_type) + + def has_pyiceberg_catalog_config(name): + calls.append(("has_pyiceberg_config", name)) + return has_pyiceberg_config + monkeypatch.setattr( tables_module, "get_tower_catalog_credentials", get_tower_catalog_credentials ) monkeypatch.setattr(tables_module, "load_vended_catalog", load_vended_catalog) monkeypatch.setattr(tables_module, "load_catalog", load_catalog) + monkeypatch.setattr( + _storage.describe_catalog_api, "sync", describe_catalog_api_sync + ) + monkeypatch.setattr( + tables_module, "_has_pyiceberg_catalog_config", has_pyiceberg_catalog_config + ) ref = tables_module.tables( "events", @@ -165,14 +202,64 @@ def load_catalog(name): if expected_source == "vend": assert ref._catalog is vend_catalog assert ref._tower_vended is True - assert calls == [ - ("vend", "default", "production", "read"), - ("load_vended_catalog", "default", "read"), - ] + assert ("vend", "default", "production", "read") in calls + assert ("load_vended_catalog", "default", "read") in calls else: assert ref._catalog is configured_catalog assert ref._tower_vended is False - assert calls == [("load_catalog", "default")] + assert ("load_catalog", "default") in calls + assert ("vend", "default", "production", "read") not in calls + + if tower_credentials is None: + assert ("describe_catalog", "default", "production") in calls + if catalog_type is None: + assert ("has_pyiceberg_config", "default") in calls + else: + assert ("describe_catalog", "default", "production") not in calls + assert ("has_pyiceberg_config", "default") not in calls + + +def test_pyiceberg_catalog_config_detects_runner_env(monkeypatch): + monkeypatch.setenv("PYICEBERG_CATALOG__S3_TABLES__URI", "https://example.com") + + assert tables_module._has_pyiceberg_catalog_config("s3-tables") is True + assert tables_module._has_pyiceberg_catalog_config("other") is False + + +def test_string_catalog_type_describe_is_cached(monkeypatch): + _storage._clear_credential_cache() + patch_tower_context(monkeypatch) + calls = [] + vend_catalogs = [] + + def describe_catalog_api_sync(name, client, environment): + calls.append(("describe_catalog", name, environment)) + return make_describe_catalog_response(name, "tower-catalog") + + def get_tower_catalog_credentials(name, environment=None, mode="read"): + calls.append(("vend", name, environment, mode)) + return make_catalog_credentials(mode) + + def load_vended_catalog(name, credentials): + catalog = FakeCatalog(credentials.mode) + vend_catalogs.append(catalog) + return catalog + + monkeypatch.setattr( + _storage.describe_catalog_api, "sync", describe_catalog_api_sync + ) + monkeypatch.setattr( + tables_module, "get_tower_catalog_credentials", get_tower_catalog_credentials + ) + monkeypatch.setattr(tables_module, "load_vended_catalog", load_vended_catalog) + + first = tables_module.tables("events", catalog="default") + second = tables_module.tables("users", catalog="default") + + assert first._catalog is vend_catalogs[0] + assert second._catalog is vend_catalogs[1] + assert calls.count(("describe_catalog", "default", "production")) == 1 + assert calls.count(("vend", "default", "production", "read")) == 2 def test_vended_table_write_lazily_escalates_and_reuses_catalog(monkeypatch): diff --git a/uv.lock b/uv.lock index 4e750275..671d0068 100644 --- a/uv.lock +++ b/uv.lock @@ -2080,7 +2080,7 @@ wheels = [ [[package]] name = "tower" -version = "0.3.68" +version = "0.3.69" source = { editable = "." } dependencies = [ { name = "attrs" },