feat: add turbo replication support and samples (#622)

* feat: add turbo replication support

* lintfix

* nother-lint-fix

* hmm weird

* i need to learn how to use lint better...

* .

* .

* how about now?

* take rpo out of constructor

* add unit tests

* add link to docs

* ensure inclusion of "rpo" in bucket._changes

* add rpo samples

* lint it

* address cathys nits

* fix a little test thing

* start to fix weirdness, creating issue to address more fully

* change it back

google/cloud/storage/bucket.py

@@ -631,6 +631,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
+"""

samples/snippets/rpo_test.py

@@ -0,0 +1,61 @@
+# Copyright 2021 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
+#
+#     http://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 uuid
+
+from google.cloud import storage
+import pytest
+
+import storage_create_bucket_turbo_replication
+import storage_get_rpo
+import storage_set_rpo_async_turbo
+import storage_set_rpo_default
+
+
+@pytest.fixture
+def dual_region_bucket():
+    """Yields a dual region bucket that is deleted after the test completes."""
+    bucket = None
+    while bucket is None or bucket.exists():
+        bucket_name = "bucket-lock-{}".format(uuid.uuid4())
+        bucket = storage.Client().bucket(bucket_name)
+        bucket.location = "NAM4"
+    bucket.create()
+    yield bucket
+    bucket.delete(force=True)
+
+
+def test_get_rpo(dual_region_bucket, capsys):
+    storage_get_rpo.get_rpo(dual_region_bucket.name)
+    out, _ = capsys.readouterr()
+    assert f"RPO for {dual_region_bucket.name} is DEFAULT." in out
+
+
+def test_set_rpo_async_turbo(dual_region_bucket, capsys):
+    storage_set_rpo_async_turbo.set_rpo_async_turbo(dual_region_bucket.name)
+    out, _ = capsys.readouterr()
+    assert f"RPO is ASYNC_TURBO for {dual_region_bucket.name}." in out
+
+
+def test_set_rpo_default(dual_region_bucket, capsys):
+    storage_set_rpo_default.set_rpo_default(dual_region_bucket.name)
+    out, _ = capsys.readouterr()
+    assert f"RPO is DEFAULT for {dual_region_bucket.name}." in out
+
+
+def test_create_bucket_turbo_replication(capsys):
+    bucket_name = "test-rpo-{}".format(uuid.uuid4())
+    storage_create_bucket_turbo_replication.create_bucket_turbo_replication(bucket_name)
+    out, _ = capsys.readouterr()
+    assert f"{bucket_name} created with RPO ASYNC_TURBO in NAM4." in out

samples/snippets/storage_create_bucket_turbo_replication.py

@@ -0,0 +1,48 @@
+#!/usr/bin/env python
+
+# Copyright 2021 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
+#
+#     http://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
+
+"""Sample that creates a new bucket with dual-region and turbo replication.
+This sample is used on this page:
+    https://cloud.google.com/storage/docs/managing-turbo-replication
+For more information, see README.md.
+"""
+
+# [START storage_create_bucket_turbo_replication]
+
+from google.cloud import storage
+from google.cloud.storage.constants import RPO_ASYNC_TURBO
+
+
+def create_bucket_turbo_replication(bucket_name):
+    """Creates dual-region bucket with turbo replication enabled."""
+    # The ID of your GCS bucket
+    # bucket_name = "my-bucket"
+
+    storage_client = storage.Client()
+    bucket = storage_client.bucket(bucket_name)
+    bucket.location = "NAM4"
+    bucket.rpo = RPO_ASYNC_TURBO
+    bucket.create()
+
+    print(f"{bucket.name} created with RPO {bucket.rpo} in {bucket.location}.")
+
+
+# [END storage_create_bucket_turbo_replication]
+
+if __name__ == "__main__":
+    create_bucket_turbo_replication(bucket_name=sys.argv[1])

samples/snippets/storage_get_rpo.py

@@ -0,0 +1,48 @@
+#!/usr/bin/env python
+
+# Copyright 2021 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
+#
+#     http://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
+
+"""Sample that gets RPO (Recovery Point Objective) of a bucket
+This sample is used on this page:
+    https://cloud.google.com/storage/docs/managing-turbo-replication
+For more information, see README.md.
+"""
+
+# [START storage_get_rpo]
+
+from google.cloud import storage
+from google.cloud.storage.constants import RPO_DEFAULT
+
+
+def get_rpo(bucket_name):
+    """Gets the RPO of the bucket"""
+    # The ID of your GCS bucket
+    # bucket_name = "my-bucket"
+
+    storage_client = storage.Client()
+    bucket = storage_client.bucket(bucket_name)
+
+    bucket.rpo = RPO_DEFAULT
+    rpo = bucket.rpo
+
+    print(f"RPO for {bucket.name} is {rpo}.")
+
+
+# [END storage_get_rpo]
+
+if __name__ == "__main__":
+    get_rpo(bucket_name=sys.argv[1])

samples/snippets/storage_set_rpo_async_turbo.py

@@ -0,0 +1,48 @@
+#!/usr/bin/env python
+
+# Copyright 2021 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
+#
+#     http://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
+
+"""Sample that sets RPO (Recovery Point Objective) to ASYNC_TURBO
+This sample is used on this page:
+    https://cloud.google.com/storage/docs/managing-turbo-replication
+For more information, see README.md.
+"""
+
+# [START storage_set_rpo_async_turbo]
+
+from google.cloud import storage
+from google.cloud.storage.constants import RPO_ASYNC_TURBO
+
+
+def set_rpo_async_turbo(bucket_name):
+    """Sets the RPO to ASYNC_TURBO, enabling the turbo replication feature"""
+    # The ID of your GCS bucket
+    # bucket_name = "my-bucket"
+
+    storage_client = storage.Client()
+    bucket = storage_client.bucket(bucket_name)
+
+    bucket.rpo = RPO_ASYNC_TURBO
+    bucket.patch()
+
+    print(f"RPO is ASYNC_TURBO for {bucket.name}.")
+
+
+# [END storage_set_rpo_async_turbo]
+
+if __name__ == "__main__":
+    set_rpo_async_turbo(bucket_name=sys.argv[1])

samples/snippets/storage_set_rpo_default.py

@@ -0,0 +1,48 @@
+#!/usr/bin/env python
+
+# Copyright 2021 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
+#
+#     http://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
+
+"""Sample that sets RPO (Recovery Point Objective) to default
+This sample is used on this page:
+    https://cloud.google.com/storage/docs/managing-turbo-replication
+For more information, see README.md.
+"""
+
+# [START storage_set_rpo_default]
+
+from google.cloud import storage
+from google.cloud.storage.constants import RPO_DEFAULT
+
+
+def set_rpo_default(bucket_name):
+    """Sets the RPO to DEFAULT, disabling the turbo replication feature"""
+    # The ID of your GCS bucket
+    # bucket_name = "my-bucket"
+
+    storage_client = storage.Client()
+    bucket = storage_client.bucket(bucket_name)
+
+    bucket.rpo = RPO_DEFAULT
+    bucket.patch()
+
+    print(f"RPO is DEFAULT for {bucket.name}.")
+
+
+# [END storage_set_rpo_default]
+
+if __name__ == "__main__":
+    set_rpo_default(bucket_name=sys.argv[1])

tests/system/test_bucket.py

@@ -885,3 +885,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"