diff --git a/google/cloud/storage/bucket.py b/google/cloud/storage/bucket.py
index 6f738976b..d071615ef 100644
--- a/google/cloud/storage/bucket.py
+++ b/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.
diff --git a/google/cloud/storage/constants.py b/google/cloud/storage/constants.py
index 2e1c1dd2a..132f4e40a 100644
--- a/google/cloud/storage/constants.py
+++ b/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
+"""
diff --git a/samples/snippets/rpo_test.py b/samples/snippets/rpo_test.py
new file mode 100644
index 000000000..d084710a9
--- /dev/null
+++ b/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
diff --git a/samples/snippets/storage_create_bucket_turbo_replication.py b/samples/snippets/storage_create_bucket_turbo_replication.py
new file mode 100644
index 000000000..68f0ba482
--- /dev/null
+++ b/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])
diff --git a/samples/snippets/storage_get_rpo.py b/samples/snippets/storage_get_rpo.py
new file mode 100644
index 000000000..29ae186fa
--- /dev/null
+++ b/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])
diff --git a/samples/snippets/storage_set_rpo_async_turbo.py b/samples/snippets/storage_set_rpo_async_turbo.py
new file mode 100644
index 000000000..10b4c67a3
--- /dev/null
+++ b/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])
diff --git a/samples/snippets/storage_set_rpo_default.py b/samples/snippets/storage_set_rpo_default.py
new file mode 100644
index 000000000..8d41b1fe0
--- /dev/null
+++ b/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])
diff --git a/tests/system/test_bucket.py b/tests/system/test_bucket.py
index 78fa135ff..dc1869d2f 100644
--- a/tests/system/test_bucket.py
+++ b/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
diff --git a/tests/unit/test_bucket.py b/tests/unit/test_bucket.py
index 8bccee19c..122233b6e 100644
--- a/tests/unit/test_bucket.py
+++ b/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"