feat: add Autoclass support and sample

google/cloud/storage/bucket.py

@@ -2660,6 +2660,50 @@ def requester_pays(self, value):
         """
         self._patch_property("billing", {"requesterPays": bool(value)})
 
+    @property
+    def autoclass_enabled(self):
+        """Whether Autoclass is enabled for this bucket.
+
+        See https://cloud.google.com/storage/docs/using-autoclass for details.
+
+        :setter: Update whether autoclass is enabled for this bucket.
+        :getter: Query whether autoclass is enabled for this bucket.
+
+        :rtype: bool
+        :returns: True if enabled, else False.
+        """
+        autoclass = self._properties.get("autoclass", {})
+        return autoclass.get("enabled", False)
+
+    @autoclass_enabled.setter
+    def autoclass_enabled(self, value):
+        """Enable or disable Autoclass at the bucket-level.
+
+        See https://cloud.google.com/storage/docs/using-autoclass for details.
+
+        :type value: convertible to boolean
+        :param value: If true, enable Autoclass for this bucket.
+                      If false, disable Autoclass for this bucket.
+
+        .. note::
+          To enable autoclass, you must set it at bucket creation time.
+          Currently, only patch requests that disable autoclass are supported.
+
+        """
+        self._patch_property("autoclass", {"enabled": bool(value)})
+
+    @property
+    def autoclass_toggle_time(self):
+        """Retrieve the toggle time when Autoclaass was last enabled or disabled for the bucket.
+        :rtype: datetime.datetime or ``NoneType``
+        :returns: point-in time at which the bucket's autoclass is toggled, or ``None`` if the property is not set locally.
+        """
+        autoclass = self._properties.get("autoclass")
+        if autoclass is not None:
+            timestamp = autoclass.get("toggleTime")
+            if timestamp is not None:
+                return _rfc3339_nanos_to_datetime(timestamp)
+
     def configure_website(self, main_page_suffix=None, not_found_page=None):
         """Configure website-related properties.
 

samples/snippets/snippets_test.py

@@ -53,6 +53,7 @@
 import storage_generate_signed_url_v2
 import storage_generate_signed_url_v4
 import storage_generate_upload_signed_url_v4
+import storage_get_autoclass
 import storage_get_bucket_labels
 import storage_get_bucket_metadata
 import storage_get_metadata
@@ -67,6 +68,7 @@
 import storage_remove_bucket_label
 import storage_remove_cors_configuration
 import storage_rename_file
+import storage_set_autoclass
 import storage_set_bucket_default_kms_key
 import storage_set_client_endpoint
 import storage_set_metadata
@@ -136,6 +138,17 @@ def test_public_bucket():
     os.environ['GOOGLE_CLOUD_PROJECT'] = original_value
 
 
+@pytest.fixture(scope="module")
+def new_bucket_obj():
+    """Yields a new bucket object that is deleted after the test completes."""
+    bucket = None
+    while bucket is None or bucket.exists():
+        bucket_name = f"storage-snippets-test-{uuid.uuid4()}"
+        bucket = storage.Client().bucket(bucket_name)
+    yield bucket
+    bucket.delete(force=True)
+
+
 @pytest.fixture
 def test_blob(test_bucket):
     """Yields a blob that is deleted after the test completes."""
@@ -408,6 +421,31 @@ def test_versioning(test_bucket, capsys):
     assert bucket.versioning_enabled is False
 
 
+def test_get_set_autoclass(new_bucket_obj, test_bucket, capsys):
+    # Test default values when Autoclass is unset
+    bucket = storage_get_autoclass.get_autoclass(test_bucket.name)
+    out, _ = capsys.readouterr()
+    assert "Autoclass enabled is set to False" in out
+    assert bucket.autoclass_toggle_time is None
+
+    # Test enabling Autoclass at bucket creation
+    new_bucket_obj.autoclass_enabled = True
+    bucket = storage.Client().create_bucket(new_bucket_obj)
+    assert bucket.autoclass_enabled is True
+
+    # Test disabling Autoclass
+    bucket = storage_set_autoclass.set_autoclass(bucket.name, False)
+    out, _ = capsys.readouterr()
+    assert "Autoclass enabled is set to False" in out
+    assert bucket.autoclass_enabled is False
+
+    # Test get Autoclass
+    bucket = storage_get_autoclass.get_autoclass(bucket.name)
+    out, _ = capsys.readouterr()
+    assert "Autoclass enabled is set to False" in out
+    assert bucket.autoclass_toggle_time is not None
+
+
 def test_bucket_lifecycle_management(test_bucket, capsys):
     bucket = storage_enable_bucket_lifecycle_management.enable_bucket_lifecycle_management(
         test_bucket

samples/snippets/storage_get_autoclass.py

@@ -0,0 +1,41 @@
+#!/usr/bin/env python
+
+# Copyright 2022 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
+
+# [START storage_get_autoclass]
+from google.cloud import storage
+
+
+def get_autoclass(bucket_name):
+    """Get the Autoclass setting for a bucket."""
+    # The ID of your GCS bucket
+    # bucket_name = "my-bucket"
+
+    storage_client = storage.Client()
+    bucket = storage_client.get_bucket(bucket_name)
+    autoclass_enabled = bucket.autoclass_enabled
+    autoclass_toggle_time = bucket.autoclass_toggle_time
+
+    print(f"Autoclass enabled is set to {autoclass_enabled} for {bucket.name} at {autoclass_toggle_time}.")
+
+    return bucket
+
+
+# [END storage_get_autoclass]
+
+if __name__ == "__main__":
+    get_autoclass(bucket_name=sys.argv[1])

samples/snippets/storage_set_autoclass.py

@@ -0,0 +1,47 @@
+#!/usr/bin/env python
+
+# Copyright 2022 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
+
+# [START storage_set_autoclass]
+from google.cloud import storage
+
+
+def set_autoclass(bucket_name, toggle):
+    """Disable Autoclass for a bucket.
+
+    Note: Only patch requests that disable autoclass are currently supported.
+    To enable autoclass, you must set it at bucket creation time.
+    """
+    # The ID of your GCS bucket
+    # bucket_name = "my-bucket"
+    # Boolean toggle - if true, enables Autoclass; if false, disables Autoclass
+    # toggle = False
+
+    storage_client = storage.Client()
+    bucket = storage_client.bucket(bucket_name)
+
+    bucket.autoclass_enabled = toggle
+    bucket.patch()
+    print(f"Autoclass enabled is set to {bucket.autoclass_enabled} for {bucket.name} at {bucket.autoclass_toggle_time}.")
+
+    return bucket
+
+
+# [END storage_set_autoclass]
+
+if __name__ == "__main__":
+    set_autoclass(bucket_name=sys.argv[1], toggle=sys.argv[2])

tests/system/test_bucket.py

@@ -996,3 +996,25 @@ def test_new_bucket_with_rpo(
     bucket_from_server = storage_client.get_bucket(bucket_name)
 
     assert bucket_from_server.rpo == constants.RPO_ASYNC_TURBO
+
+
+def test_new_bucket_with_autoclass(
+    storage_client,
+    buckets_to_delete,
+):
+    # Autoclass can be enabled/disabled via bucket create
+    bucket_name = _helpers.unique_name("new-w-autoclass")
+    bucket_obj = storage_client.bucket(bucket_name)
+    bucket_obj.autoclass_enabled = True
+    bucket = storage_client.create_bucket(bucket_obj)
+    previous_toggle_time = bucket.autoclass_toggle_time
+    buckets_to_delete.append(bucket)
+
+    assert bucket.autoclass_enabled is True
+
+    # Autoclass can be enabled/disabled via bucket patch
+    bucket.autoclass_enabled = False
+    bucket.patch()
+
+    assert bucket.autoclass_enabled is False
+    assert bucket.autoclass_toggle_time != previous_toggle_time

tests/unit/test_bucket.py

@@ -2644,6 +2644,37 @@ def test_rpo_getter_and_setter(self):
         self.assertIn("rpo", bucket._changes)
         self.assertEqual(bucket.rpo, RPO_DEFAULT)
 
+    def test_autoclass_enabled_getter_and_setter(self):
+        properties = {"autoclass": {"enabled": True}}
+        bucket = self._make_one(properties=properties)
+        self.assertTrue(bucket.autoclass_enabled)
+        bucket.autoclass_enabled = False
+        self.assertIn("autoclass", bucket._changes)
+        self.assertFalse(bucket.autoclass_enabled)
+
+    def test_autoclass_toggle_time_missing(self):
+        bucket = self._make_one()
+        self.assertIsNone(bucket.autoclass_toggle_time)
+
+        properties = {"autoclass": {}}
+        bucket = self._make_one(properties=properties)
+        self.assertIsNone(bucket.autoclass_toggle_time)
+
+    def test_autoclass_toggle_time(self):
+        import datetime
+        from google.cloud._helpers import _datetime_to_rfc3339
+        from google.cloud._helpers import UTC
+
+        effective_time = datetime.datetime.utcnow().replace(tzinfo=UTC)
+        properties = {
+            "autoclass": {
+                "enabled": True,
+                "toggleTime": _datetime_to_rfc3339(effective_time),
+            }
+        }
+        bucket = self._make_one(properties=properties)
+        self.assertEqual(bucket.autoclass_toggle_time, effective_time)
+
     def test_get_logging_w_prefix(self):
         NAME = "name"
         LOG_BUCKET = "logs"