feat: add turbo replication support and samples

google/cloud/storage/bucket.py

@@ -55,6 +55,8 @@
 from google.cloud.storage.constants import REGIONAL_LEGACY_STORAGE_CLASS
 from google.cloud.storage.constants import REGION_LOCATION_TYPE
 from google.cloud.storage.constants import STANDARD_STORAGE_CLASS
+from google.cloud.storage.constants import RPO_DEFAULT
+from google.cloud.storage.constants import RPO_ASYNC_TURBO
 from google.cloud.storage.notification import BucketNotification
 from google.cloud.storage.notification import NONE_PAYLOAD_FORMAT
 from google.cloud.storage.retry import DEFAULT_RETRY
@@ -633,6 +635,25 @@ def _set_properties(self, value):
         self._label_removals.clear()
         return super(Bucket, self)._set_properties(value)
 
+    @property
+    def rpo(self):
+        """Get the RPO (Recovery Point Objective) of this bucket
+
+        "ASYNC_TURBO" or "DEFAULT"
+        :rtype: str
+        """
+        return self._properties.get("rpo")
+
+    @rpo.setter
+    def rpo(self, value):
+        """
+        Set the RPO (Recovery Point Objective) of this bucket.
+
+        :type value: str
+        :param value: "ASYNC_TURBO" or "DEFAULT"
+        """
+        self._patch_property("rpo", value)
+
     @property
     def user_project(self):
         """Project ID to be billed for API requests made via this bucket.

google/cloud/storage/client.py

@@ -832,6 +832,7 @@ def create_bucket(
         location=None,
         predefined_acl=None,
         predefined_default_object_acl=None,
+        rpo=None,
         timeout=_DEFAULT_TIMEOUT,
         retry=DEFAULT_RETRY,
     ):
@@ -953,6 +954,9 @@ def create_bucket(
         if location is not None:
             properties["location"] = location
 
+        if rpo is not None:
+            properties["rpo"] = rpo
+
         api_response = self._post_resource(
             "/b",
             properties,

google/cloud/storage/constants.py

@@ -117,3 +117,15 @@
 
 See: https://cloud.google.com/storage/docs/public-access-prevention
 """
+
+RPO_ASYNC_TURBO = "ASYNC_TURBO"
+"""Turbo Replication RPO
+
+See: https://cloud.google.com/storage/docs/managing-turbo-replication
+"""
+
+RPO_DEFAULT = "DEFAULT"
+"""Default RPO
+
+See: https://cloud.google.com/storage/docs/managing-turbo-replication
+"""

tests/system/test_bucket.py

@@ -886,3 +886,21 @@ def test_new_bucket_created_w_enforced_pap(
         constants.PUBLIC_ACCESS_PREVENTION_INHERITED,
     ]
     assert not bucket.iam_configuration.uniform_bucket_level_access_enabled
+
+def test_new_bucket_with_rpo(
+    storage_client, buckets_to_delete, blobs_to_delete,
+):
+    from google.cloud.storage import constants
+
+    bucket_name = _helpers.unique_name("new-w-turbo-replication")
+    bucket = storage_client.create_bucket(bucket_name, location="NAM4", rpo=constants.RPO_ASYNC_TURBO)
+    buckets_to_delete.append(bucket)
+
+    assert bucket.rpo == constants.RPO_ASYNC_TURBO
+
+    bucket.rpo = constants.RPO_DEFAULT
+    bucket.patch()
+
+    bucket_from_server = storage_client.get_bucket(bucket_name)
+
+    assert bucket_from_server.rpo == constants.RPO_DEFAULT