Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 9 additions & 1 deletion CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,13 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0

## [Unreleased]

### Fixed

- `tilebox-workflows`: Restore `GoogleStorageCache` support for Google SDK bucket objects, preserving their
configured client, credentials, and request settings. Bucket-name strings continue to use obstore.

## [0.63.0] - 2026-10-03

### Added

- Support Python 3.14 across all packages.
Expand Down Expand Up @@ -568,7 +575,8 @@ the first client that does not cache data (since it's already on the local file
- Released under the [MIT](https://opensource.org/license/mit) license.
- Released packages: `tilebox-datasets`, `tilebox-workflows`, `tilebox-storage`, `tilebox-grpc`

[Unreleased]: https://github.com/tilebox/tilebox-python/compare/v0.61.0...HEAD
[Unreleased]: https://github.com/tilebox/tilebox-python/compare/v0.63.0...HEAD
[0.63.0]: https://github.com/tilebox/tilebox-python/compare/v0.62.0...v0.63.0
[0.61.0]: https://github.com/tilebox/tilebox-python/compare/v0.60.0...v0.61.0
[0.60.0]: https://github.com/tilebox/tilebox-python/compare/v0.59.0...v0.60.0
[0.59.0]: https://github.com/tilebox/tilebox-python/compare/v0.58.0...v0.59.0
Expand Down
2 changes: 2 additions & 0 deletions tilebox-workflows/pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -50,6 +50,8 @@ geospatial = [
[dependency-groups]
dev = [
"boto3-stubs[s3]>=1.40.2",
"google-cloud-storage>=3.2.0",
"grpcio-status>=1.78.0",
"hypothesis>=6.112.1",
"ipywidgets>=8.1.7",
"odc-geo>=0.5",
Expand Down
7 changes: 6 additions & 1 deletion tilebox-workflows/tests/test_cache.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
from datetime import timedelta
from pathlib import Path
from typing import assert_type
from unittest.mock import MagicMock, patch

import boto3
Expand Down Expand Up @@ -170,11 +171,15 @@ def test_google_storage_credentials_and_groups() -> None:
):
cache = GoogleStorageCache("cache-bucket")
first = cache.group("folder")
assert_type(cache.bucket, str)
assert_type(first.bucket, str)
assert_type(first, GoogleStorageCache[str])
first["one"] = b"first"
cache.group("folder2")["two"] = b"second"
assert list(first) == ["one"]
assert first["one"] == b"first"
assert isinstance(first, ObstoreCache)
assert isinstance(first, GoogleStorageCache)
assert first.bucket == "cache-bucket"
assert first.store is store
assert store.get("jobs/folder/one").bytes() == b"first"
default.assert_called_once_with(scopes=["https://www.googleapis.com/auth/devstorage.read_write"])
Expand Down
96 changes: 96 additions & 0 deletions tilebox-workflows/tests/test_google_cache_compatibility.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,96 @@
from collections.abc import Iterator
from typing import assert_type
from unittest.mock import MagicMock, patch
from urllib.parse import parse_qs, urlsplit

import pytest
from google.api_core.exceptions import Forbidden, NotFound, ServiceUnavailable
from google.auth.credentials import AnonymousCredentials
from google.cloud.storage import Blob, Bucket, Client
from google.oauth2.credentials import Credentials
from requests import Response

from tilebox.workflows.cache import GoogleStorageCache


@pytest.fixture(params=[False, True], ids=["authenticated", "anonymous"])
def google_bucket(request: pytest.FixtureRequest) -> Iterator[Bucket]:
credentials = AnonymousCredentials() if request.param else Credentials(token="configured-token") # noqa: S106
client = Client(
project="configured-project",
credentials=credentials,
_http=MagicMock(),
client_options={"api_endpoint": "https://custom.example.test"},
)
with (
patch("google.auth.default", side_effect=AssertionError("Bucket credentials must not be replaced by ADC")),
patch("obstore.store.GCSStore", side_effect=AssertionError("Bucket configuration must stay with its client")),
):
yield client.bucket("configured-bucket", user_project="billing-project")


def test_google_bucket_read_preserves_client(google_bucket: Bucket) -> None:
# Bucket reads and nested groups retain credentials, custom transport, endpoint, and requester-pays billing.
credentials = google_bucket.client._credentials
transport = google_bucket.client._http
response = Response()
response.status_code = 200
response._content = b"configured client content"
response._content_consumed = True
response.raw = MagicMock(headers={})
response.headers["content-length"] = str(len(response.content))
transport.request.return_value = response
root = GoogleStorageCache(google_bucket)
cache = root.group("folder").group("nested")
assert_type(root.bucket, Bucket)
assert_type(cache.bucket, Bucket)
assert_type(cache, GoogleStorageCache[Bucket])
assert isinstance(cache, GoogleStorageCache)
assert cache.bucket is root.bucket is google_bucket
assert cache["file.txt"] == b"configured client content"
assert google_bucket.client._credentials is credentials
# Newer Google clients may also fetch bucket metadata in a background thread.
downloads = [call for call in transport.request.call_args_list if call.args and "/download/" in call.args[1]]
assert len(downloads) == 1
method, url = downloads[0].args
assert method == "GET"
parsed = urlsplit(url)
assert parsed.netloc == "custom.example.test"
assert parsed.path.endswith("/configured-bucket/o/jobs%2Ffolder%2Fnested%2Ffile.txt")
assert parse_qs(parsed.query)["userProject"] == ["billing-project"]


@pytest.mark.parametrize("prefix", ["jobs", ""])
def test_google_bucket_operations(google_bucket: Bucket, prefix: str) -> None:
# Bucket caches preserve writes, membership, deletion, and delimiter-based listing for each group.
cache = GoogleStorageCache(google_bucket, prefix).group("folder")
path = f"{prefix}/folder" if prefix else "folder"
with (
patch.object(Blob, "upload_from_file", autospec=True) as upload,
patch.object(Blob, "exists", autospec=True, return_value=True) as exists,
patch.object(Blob, "delete", autospec=True) as delete,
patch.object(google_bucket, "list_blobs", return_value=[Blob(f"{path}/one", google_bucket)]) as listing,
):
cache["one"] = b"saved value"
blob, stream = upload.call_args.args
assert blob.bucket is google_bucket
assert blob.name == f"{path}/one"
assert stream.getvalue() == b"saved value"
assert "one" in cache
assert exists.call_args.args[0].name == f"{path}/one"
assert list(cache) == ["one"]
listing.assert_called_once_with(prefix=f"{path}/", delimiter="/")
del cache["one"]
assert delete.call_args.args[0].name == f"{path}/one"


@pytest.mark.parametrize("error_type", [NotFound, Forbidden, ServiceUnavailable])
def test_google_bucket_read_errors(google_bucket: Bucket, error_type: type[Exception]) -> None:
# Only missing objects become KeyError; configured-client permission and backend errors propagate unchanged.
error = error_type("read failed")
with patch.object(Blob, "download_as_bytes", side_effect=error):
expected = KeyError if error_type is NotFound else error_type
with pytest.raises(expected) as raised:
GoogleStorageCache(google_bucket).group("nested")["file"]
if error_type is not NotFound:
assert raised.value is error
82 changes: 77 additions & 5 deletions tilebox-workflows/tilebox/workflows/cache.py
Original file line number Diff line number Diff line change
@@ -1,10 +1,12 @@
import contextlib
from abc import ABC, abstractmethod
from collections.abc import Iterator
from collections.abc import Iterable, Iterator
from copy import copy
from io import BytesIO
from pathlib import Path
from pathlib import PurePosixPath as ObjectPath
from threading import RLock
from typing import TYPE_CHECKING, Any
from typing import TYPE_CHECKING, Any, Generic, Protocol, Self, TypeVar

if TYPE_CHECKING:
from obstore.store import ObjectStore
Expand Down Expand Up @@ -257,14 +259,37 @@ def group(self, key: str) -> "LocalFileSystemCache":
return LocalFileSystemCache(self.root / key)


class GoogleStorageCache(ObstoreCache):
def __init__(self, bucket: str, prefix: str | ObjectPath = "jobs") -> None:
class _GoogleBlob(Protocol):
@property
def name(self) -> str: ...
def exists(self) -> bool: ...
def upload_from_file(self, stream: BytesIO, /) -> None: ...
def download_as_bytes(self) -> bytes: ...
def delete(self) -> None: ...


class _GoogleBucket(Protocol):
def blob(self, name: str, /) -> _GoogleBlob: ...
def list_blobs(self, *, prefix: str, delimiter: str) -> Iterable[_GoogleBlob]: ...


_BucketT = TypeVar("_BucketT", bound=str | _GoogleBucket)


class GoogleStorageCache(ObstoreCache, Generic[_BucketT]):
def __init__(self, bucket: _BucketT, prefix: str | ObjectPath = "jobs") -> None:
"""A cache implementation that stores data in Google Cloud Storage.

Args:
bucket: The Google Cloud Storage bucket name. Credentials come from Google Auth.
bucket: A bucket name or Google SDK Bucket. Names use obstore and Google Auth;
Bucket objects keep their existing client and its configuration.
prefix: A path prefix to append to all objects stored in the cache. Defaults to "jobs".
"""
self.bucket = bucket
if not isinstance(bucket, str):
self.prefix = ObjectPath(prefix)
return

import google.auth # noqa: PLC0415
from obstore.auth.google import GoogleCredentialProvider # noqa: PLC0415
from obstore.store import GCSStore # noqa: PLC0415
Expand All @@ -273,6 +298,53 @@ def __init__(self, bucket: str, prefix: str | ObjectPath = "jobs") -> None:
store = GCSStore(bucket, credential_provider=GoogleCredentialProvider(credentials=credentials))
super().__init__(store, prefix)

def __contains__(self, key: str) -> bool:
if isinstance(self.bucket, str):
return super().__contains__(key)
return self.bucket.blob(str(self.prefix / key)).exists()

def __setitem__(self, key: str, value: bytes) -> None:
if isinstance(self.bucket, str):
super().__setitem__(key, value)
else:
self.bucket.blob(str(self.prefix / key)).upload_from_file(BytesIO(value))

def __getitem__(self, key: str) -> bytes:
if isinstance(self.bucket, str):
return super().__getitem__(key)

from google.api_core.exceptions import NotFound # noqa: PLC0415

try:
return self.bucket.blob(str(self.prefix / key)).download_as_bytes()
except NotFound:
raise KeyError(f"{key} is not cached!") from None

def __delitem__(self, key: str) -> None:
if isinstance(self.bucket, str):
super().__delitem__(key)
return

from google.api_core.exceptions import NotFound # noqa: PLC0415

try:
self.bucket.blob(str(self.prefix / key)).delete()
except NotFound:
raise KeyError(f"{key} is not cached!") from None

def __iter__(self) -> Iterator[str]:
if isinstance(self.bucket, str):
yield from super().__iter__()
return
prefix = "" if self.prefix == ObjectPath(".") else str(self.prefix) + "/"
for blob in self.bucket.list_blobs(prefix=prefix, delimiter="/"):
yield str(ObjectPath(blob.name).relative_to(self.prefix))

def group(self, key: str) -> Self:
group = copy(self)
group.prefix = self.prefix / key
return group


class AmazonS3Cache(ObstoreCache):
def __init__(self, bucket: str, prefix: str | ObjectPath = "jobs", *, region: str | None = None) -> None:
Expand Down
Loading
Loading