feat: add turbo replication support and samples

google/cloud/storage/bucket.py

@@ -633,6 +633,29 @@ 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
+
+        See: https://cloud.google.com/storage/docs/managing-turbo-replication
+
+        "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.
+
+        See: https://cloud.google.com/storage/docs/managing-turbo-replication
+
+        :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/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,22 @@ 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")
+    buckets_to_delete.append(bucket)
+
+    assert bucket.rpo == constants.RPO_DEFAULT
+
+    bucket.rpo = constants.RPO_ASYNC_TURBO
+    bucket.patch()
+
+    bucket_from_server = storage_client.get_bucket(bucket_name)
+
+    assert bucket_from_server.rpo == constants.RPO_ASYNC_TURBO

tests/unit/test_bucket.py

@@ -25,6 +25,8 @@
 from google.cloud.storage.constants import PUBLIC_ACCESS_PREVENTION_ENFORCED
 from google.cloud.storage.constants import PUBLIC_ACCESS_PREVENTION_INHERITED
 from google.cloud.storage.constants import PUBLIC_ACCESS_PREVENTION_UNSPECIFIED
+from google.cloud.storage.constants import RPO_DEFAULT
+from google.cloud.storage.constants import RPO_ASYNC_TURBO
 
 
 def _create_signing_credentials():
@@ -2476,6 +2478,14 @@ def test_location_type_getter_set(self):
         bucket = self._make_one(properties=properties)
         self.assertEqual(bucket.location_type, REGION_LOCATION_TYPE)
 
+    def test_rpo_getter_and_setter(self):
+        bucket = self._make_one()
+        bucket.rpo = RPO_ASYNC_TURBO
+        self.assertEqual(bucket.rpo, RPO_ASYNC_TURBO)
+        bucket.rpo = RPO_DEFAULT
+        self.assertIn("rpo", bucket._changes)
+        self.assertEqual(bucket.rpo, RPO_DEFAULT)
+
     def test_get_logging_w_prefix(self):
         NAME = "name"
         LOG_BUCKET = "logs"