feat(storagecontrol): add Rapid Cache samples

storagecontrol/conftest.py

@@ -13,12 +13,18 @@
 # limitations under the License.
 
 import os
+import time
 import uuid
 
+from google.api_core import exceptions
+from google.api_core import retry
 from google.cloud import storage
+from google.cloud import storage_control_v2
 
 import pytest
 
+import logging
+
 
 @pytest.fixture(scope="module")
 def project_id() -> str:
@@ -66,6 +72,83 @@ def hns_enabled_bucket(project_id: str, bucket_name: str) -> storage.Bucket:
     bucket.delete(force=True)
 
 
+@pytest.fixture(scope="module")
+def rapid_cache_zone() -> str:
+    yield os.environ.get("RAPID_CACHE_ZONE", "us-central1-a")
+
+
+@pytest.fixture(scope="function")
+def regional_hns_enabled_bucket(
+    project_id: str, bucket_name: str, rapid_cache_zone: str
+) -> storage.Bucket:
+    """
+    Yields and auto-cleans up a regional HNS enabled bucket located in the
+    region of the Rapid Cache zone.
+    """
+
+    storage_client = storage.Client(project=project_id)
+    bucket = storage_client.bucket(bucket_name)
+    bucket.iam_configuration.uniform_bucket_level_access_enabled = True
+    bucket.hierarchical_namespace_enabled = True
+    region = rapid_cache_zone.rsplit("-", 1)[0]
+    bucket = storage_client.create_bucket(bucket, location=region)
+    # Wait for the new bucket's metadata to propagate before creating a Rapid
+    # Cache on it.
+    time.sleep(20)
+
+    yield bucket
+
+    # A bucket cannot be deleted until its disabled Rapid Cache has been fully
+    # removed, which can take more than an hour. Cleanup is best-effort: a
+    # short retry, after which the bucket is left behind for later cleanup.
+    try:
+        retry.Retry(
+            predicate=retry.if_exception_type(exceptions.Conflict),
+            initial=10.0,
+            maximum=30.0,
+            timeout=120.0,
+        )(bucket.delete)(force=True)
+    except exceptions.GoogleAPIError as e:
+        logging.warning("Could not delete bucket %s: %s", bucket.name, e)
+
+
+@pytest.fixture(scope="function")
+def rapid_cache_id(
+    regional_hns_enabled_bucket: storage.Bucket, rapid_cache_zone: str
+) -> str:
+    """
+    Yields the ID of a Rapid Cache for the regional HNS enabled bucket. If the
+    test created the Rapid Cache and left it active, it is disabled on
+    teardown so that the bucket can be deleted once the cache is removed.
+    """
+
+    # The Rapid Cache ID is the same as its zone.
+    cache_id = rapid_cache_zone
+
+    yield cache_id
+
+    storage_control_client = storage_control_v2.StorageControlClient()
+    bucket_path = (
+        f"{storage_control_client.common_project_path('_')}"
+        f"/buckets/{regional_hns_enabled_bucket.name}"
+    )
+    cache_name = storage_control_client.rapid_cache_path(
+        project="_", bucket=regional_hns_enabled_bucket.name, rapid_cache=cache_id
+    )
+    # Cleanup is best-effort: errors are logged instead of failing the test.
+    try:
+        caches = storage_control_client.list_rapid_caches(parent=bucket_path)
+        cache_active = any(
+            cache.zone == cache_id and cache.state != "disabled" for cache in caches
+        )
+        if cache_active:
+            # Disabling is a long-running operation that completes only once
+            # the cache is fully removed, so teardown does not wait for it.
+            storage_control_client.disable_rapid_cache(name=cache_name)
+    except exceptions.GoogleAPIError as e:
+        logging.warning("Could not disable Rapid Cache %s: %s", cache_name, e)
+
+
 @pytest.fixture(scope="function")
 def ubla_enabled_bucket(project_id: str, bucket_name: str) -> storage.Bucket:
     """

storagecontrol/create_rapid_cache.py

@@ -0,0 +1,67 @@
+# Copyright 2026 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#    https://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import sys
+
+# [START storage_control_create_rapid_cache]
+from google.cloud import storage_control_v2
+
+
+def create_rapid_cache(bucket_name: str, zone: str) -> None:
+    # The ID of your GCS bucket
+    # bucket_name = "your-unique-bucket-name"
+
+    # The zone in which to create the Rapid Cache. The zone must be in the
+    # same region as the bucket.
+    # zone = "us-central1-a"
+
+    storage_control_client = storage_control_v2.StorageControlClient()
+    # The storage bucket path uses the global access pattern, in which the "_"
+    # denotes this bucket exists in the global namespace.
+    project_path = storage_control_client.common_project_path("_")
+    bucket_path = f"{project_path}/buckets/{bucket_name}"
+    # The ID of the Rapid Cache must be the same as its zone.
+    cache_id = zone
+    rapid_cache_path = storage_control_client.rapid_cache_path(
+        project="_", bucket=bucket_name, rapid_cache=cache_id
+    )
+
+    # Only the required fields are set. Optional fields, such as the time to
+    # live (TTL) for cache entries, use server defaults and can be changed
+    # later by updating the Rapid Cache.
+    # Rapid Cache Ultra requires a bucket with hierarchical namespace enabled.
+    rapid_cache = storage_control_v2.RapidCache(
+        name=rapid_cache_path,
+        zone=zone,
+        cache_type="rapid-cache-ultra",
+    )
+    request = storage_control_v2.CreateRapidCacheRequest(
+        parent=bucket_path,
+        rapid_cache=rapid_cache,
+    )
+
+    # Creating a Rapid Cache is a long-running operation.
+    operation = storage_control_client.create_rapid_cache(request=request)
+    response = operation.result()
+
+    print(f"Created Rapid Cache: {response.name}")
+    print(f"State: {response.state}")
+    print(f"Cache type: {response.cache_type}")
+
+
+# [END storage_control_create_rapid_cache]
+
+
+if __name__ == "__main__":
+    create_rapid_cache(bucket_name=sys.argv[1], zone=sys.argv[2])

storagecontrol/disable_rapid_cache.py

@@ -0,0 +1,57 @@
+# Copyright 2026 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#    https://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import sys
+
+# [START storage_control_disable_rapid_cache]
+from google.cloud import storage_control_v2
+
+
+def disable_rapid_cache(
+    bucket_name: str, cache_id: str, timeout: float | None = None
+) -> None:
+    # The ID of your GCS bucket
+    # bucket_name = "your-unique-bucket-name"
+
+    # The ID of the Rapid Cache, which is the same as its zone
+    # cache_id = "us-central1-a"
+
+    # (Optional) The maximum time to wait for the operation, in seconds.
+    # None waits until the operation completes.
+    # timeout = None
+
+    storage_control_client = storage_control_v2.StorageControlClient()
+    # The storage bucket path uses the global access pattern, in which the "_"
+    # denotes this bucket exists in the global namespace.
+    rapid_cache_path = storage_control_client.rapid_cache_path(
+        project="_", bucket=bucket_name, rapid_cache=cache_id
+    )
+
+    request = storage_control_v2.DisableRapidCacheRequest(name=rapid_cache_path)
+
+    # Disabling a Rapid Cache is a long-running operation. It completes only
+    # after the cache is fully removed, which can take more than an hour.
+    # This sample waits for the operation to complete, or until the timeout.
+    operation = storage_control_client.disable_rapid_cache(request=request)
+    response = operation.result(timeout=timeout)
+
+    print(f"Disabled Rapid Cache: {response.name}")
+    print(f"State: {response.state}")
+
+
+# [END storage_control_disable_rapid_cache]
+
+
+if __name__ == "__main__":
+    disable_rapid_cache(bucket_name=sys.argv[1], cache_id=sys.argv[2])

storagecontrol/get_rapid_cache.py

@@ -0,0 +1,48 @@
+# Copyright 2026 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#    https://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import sys
+
+# [START storage_control_get_rapid_cache]
+from google.cloud import storage_control_v2
+
+
+def get_rapid_cache(bucket_name: str, cache_id: str) -> None:
+    # The ID of your GCS bucket
+    # bucket_name = "your-unique-bucket-name"
+
+    # The ID of the Rapid Cache, which is the same as its zone
+    # cache_id = "us-central1-a"
+
+    storage_control_client = storage_control_v2.StorageControlClient()
+    # The storage bucket path uses the global access pattern, in which the "_"
+    # denotes this bucket exists in the global namespace.
+    rapid_cache_path = storage_control_client.rapid_cache_path(
+        project="_", bucket=bucket_name, rapid_cache=cache_id
+    )
+
+    request = storage_control_v2.GetRapidCacheRequest(name=rapid_cache_path)
+    response = storage_control_client.get_rapid_cache(request=request)
+
+    print(f"Got Rapid Cache: {response.name}")
+    print(f"Zone: {response.zone}")
+    print(f"State: {response.state}")
+    print(f"Cache type: {response.cache_type}")
+
+
+# [END storage_control_get_rapid_cache]
+
+
+if __name__ == "__main__":
+    get_rapid_cache(bucket_name=sys.argv[1], cache_id=sys.argv[2])

storagecontrol/list_rapid_caches.py

@@ -0,0 +1,45 @@
+# Copyright 2026 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#    https://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import sys
+
+# [START storage_control_list_rapid_caches]
+from google.cloud import storage_control_v2
+
+
+def list_rapid_caches(bucket_name: str) -> None:
+    # The ID of your GCS bucket
+    # bucket_name = "your-unique-bucket-name"
+
+    storage_control_client = storage_control_v2.StorageControlClient()
+    # The storage bucket path uses the global access pattern, in which the "_"
+    # denotes this bucket exists in the global namespace.
+    project_path = storage_control_client.common_project_path("_")
+    bucket_path = f"{project_path}/buckets/{bucket_name}"
+
+    request = storage_control_v2.ListRapidCachesRequest(parent=bucket_path)
+
+    # The returned pager handles pagination automatically.
+    page_result = storage_control_client.list_rapid_caches(request=request)
+    for rapid_cache in page_result:
+        print(rapid_cache.name)
+
+    print(f"Listed Rapid Caches in bucket {bucket_name}")
+
+
+# [END storage_control_list_rapid_caches]
+
+
+if __name__ == "__main__":
+    list_rapid_caches(bucket_name=sys.argv[1])