Add Google Cloud Storage support for external payloads

README.md

@@ -556,6 +556,7 @@ Some things to note about driver selection:
 
 ###### Built-in Drivers
 
+- **[GCS Storage Driver](temporalio/contrib/gcp/gcsdriver/)**: ⚠️ **Experimental** ⚠️ Google Cloud Storage driver. Ships with a gcloud-aio-storage client adapter, or bring your own by subclassing `GCSStorageDriverClient`.
 - **[S3 Storage Driver](temporalio/contrib/aws/s3driver/)**: ⚠️ **Experimental** ⚠️ Amazon S3 driver. Ships with an aioboto3 client, or bring your own by subclassing `S3StorageDriverClient`.
 
 ###### Custom Drivers

changelog/added/bouncing-cloudberry.md

@@ -0,0 +1 @@
+Add an experimental Google Cloud Storage external payload storage driver with an optional async `gcloud-aio-storage` adapter.

pyproject.toml

@@ -51,6 +51,7 @@ cloud-run-worker-otel = [
   "protobuf<7",
 ]
 aioboto3 = ["aioboto3>=10.4.0", "types-aioboto3[s3]>=10.4.0"]
+gcloud-aio-storage = ["gcloud-aio-storage>=9.6.4,<10"]
 google-genai = ["temporalio-google-genai[mcp]>=0.1.0"]
 strands-agents = ["temporalio-strands-agents>=0.1.0"]
 

temporalio/contrib/gcp/gcsdriver/README.md

@@ -0,0 +1,91 @@
+# Google Cloud Storage Driver
+
+> ⚠️ **This package is experimental.** ⚠️
+
+`GCSStorageDriver` stores and retrieves Temporal payloads in Google Cloud Storage.
+
+Install the async GCS client:
+
+```bash
+python -m pip install "temporalio[gcloud-aio-storage]"
+```
+
+```python
+import dataclasses
+
+from gcloud.aio.storage import Storage
+
+from temporalio.client import Client
+from temporalio.contrib.gcp.gcsdriver import GCSStorageDriver
+from temporalio.contrib.gcp.gcsdriver.gcloud_aio import new_gcloud_aio_client
+from temporalio.converter import DataConverter, ExternalStorage
+
+async with Storage() as storage:
+    driver = GCSStorageDriver(
+        client=new_gcloud_aio_client(storage),
+        bucket="my-temporal-payloads",
+    )
+    client = await Client.connect(
+        "localhost:7233",
+        data_converter=dataclasses.replace(
+            DataConverter.default,
+            external_storage=ExternalStorage(drivers=[driver]),
+        ),
+    )
+    # Run clients and workers inside this context so their storage client stays open.
+```
+
+Configure credentials, timeouts, and retries through the `Storage` client. See the
+[gcloud-aio documentation](https://talkiq.github.io/gcloud-aio/) for details.
+The adapter does not add retries, and interrupted uploads must restart.
+
+## Configuration
+
+- Configure this driver on every Temporal client and worker that stores or retrieves
+  its payloads, using the same `driver_name` (default: `gcp.gcsdriver`).
+- `bucket` accepts a name or a function taking `(context, payload)` and returning
+  a bucket name. Downloads use the bucket recorded at upload time.
+- The bucket must already exist. Use a flat namespace bucket and credentials with
+  `storage.objects.get` and `storage.objects.create` permissions.
+- `ExternalStorage.payload_size_threshold` defaults to 256 KiB. The driver accepts
+  serialized payloads up to `max_payload_size` (default: 50 MiB).
+- Retain objects for as long as workflow histories may need them, including replay.
+  The driver leaves deletion and lifecycle policies to the application.
+
+To use another async GCS client, implement `GCSStorageDriverClient` with
+`object_exists`, `put_object`, and `get_object`. `put_object` must leave an existing
+object unchanged.
+
+## Object names and integrity
+
+Objects store the full Temporal payload, including metadata. Names include the
+namespace, available workflow or activity identifiers, and a SHA-256 hash:
+
+```text
+v0/ns/{namespace}/wt/{workflow-type}/wi/{workflow-id}/ri/{run-id}/d/sha256/{hex-digest}
+v0/ns/{namespace}/at/{activity-type}/ai/{activity-id}/ri/{run-id}/d/sha256/{hex-digest}
+v0/d/sha256/{hex-digest}
+```
+
+The last form is used when no target is available. Empty or missing fields use
+`null`, including an empty namespace on a workflow or activity target. Names
+preserve Unicode and escape characters that interfere with
+[GCS tools](https://cloud.google.com/storage/docs/objects#naming), including `/`
+and `%` within individual fields.
+
+Names over 1,024 UTF-8 bytes fall back to
+`v0/ns/{namespace}/d/sha256/{hex-digest}`, retaining the namespace for access policies.
+Identical payload bytes in the same context reuse an object. Uploads never
+overwrite existing objects, and downloads verify the hash before decoding.
+
+## Testing
+
+Run the tests:
+
+```bash
+poe test -s tests/contrib/gcp/gcsdriver
+```
+
+Pytest downloads and caches [fake-gcs-server](https://github.com/fsouza/fake-gcs-server)
+in `.pytest_cache`, starts it locally, and cleans up after the tests. To use a
+preinstalled binary, put `fake-gcs-server` on `PATH`.

temporalio/contrib/gcp/gcsdriver/__init__.py

@@ -0,0 +1,13 @@
+"""Google Cloud Storage driver for Temporal external storage.
+
+.. warning::
+    This API is experimental.
+"""
+
+from temporalio.contrib.gcp.gcsdriver._client import GCSStorageDriverClient
+from temporalio.contrib.gcp.gcsdriver._driver import GCSStorageDriver
+
+__all__ = [
+    "GCSStorageDriverClient",
+    "GCSStorageDriver",
+]

temporalio/contrib/gcp/gcsdriver/_client.py

@@ -0,0 +1,42 @@
+"""GCS storage driver client abstraction for the GCS storage driver.
+
+.. warning::
+    This API is experimental.
+"""
+
+from __future__ import annotations
+
+from abc import ABC, abstractmethod
+from collections.abc import Mapping
+
+
+class GCSStorageDriverClient(ABC):
+    """Abstract base class for GCS object operations.
+
+    Implementations must support ``object_exists``, ``put_object``, and
+    ``get_object``. Resumable uploads (if needed) are an internal concern of
+    each implementation. Uploads must leave existing objects unchanged.
+
+    .. warning::
+        This API is experimental.
+    """
+
+    @abstractmethod
+    async def put_object(self, *, bucket: str, object_name: str, data: bytes) -> None:
+        """Upload *data* if absent at the given GCS *bucket* and *object_name*."""
+
+    @abstractmethod
+    async def object_exists(self, *, bucket: str, object_name: str) -> bool:
+        """Return ``True`` if an object exists at the given *bucket* and *object_name*."""
+
+    @abstractmethod
+    async def get_object(self, *, bucket: str, object_name: str) -> bytes:
+        """Download and return the bytes stored at the given GCS *bucket* and *object_name*."""
+
+    def describe(self) -> Mapping[str, str]:
+        """Return client-specific diagnostic metadata (e.g. region, credentials
+        source) that the driver appends to error messages. Implementations may
+        override this to surface configuration that is useful for debugging
+        common misconfigurations. Returns an empty mapping by default.
+        """
+        return {}