feat: add Autoclass v2.1 support (#1117)

* feat: add Autoclass v2.1 support

* update tests and coverage

* update samples with v2.1 additions

* fix lint

* update samples

Author: cojenco

google/cloud/storage/bucket.py

@@ -2689,13 +2689,10 @@ def autoclass_enabled(self, value):
         :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)})
+        autoclass = self._properties.get("autoclass", {})
+        autoclass["enabled"] = bool(value)
+        self._patch_property("autoclass", autoclass)
 
     @property
     def autoclass_toggle_time(self):
@@ -2709,6 +2706,48 @@ def autoclass_toggle_time(self):
             if timestamp is not None:
                 return _rfc3339_nanos_to_datetime(timestamp)
 
+    @property
+    def autoclass_terminal_storage_class(self):
+        """The storage class that objects in an Autoclass bucket eventually transition to if
+        they are not read for a certain length of time. Valid values are NEARLINE and ARCHIVE.
+
+        See https://cloud.google.com/storage/docs/using-autoclass for details.
+
+        :setter: Set the terminal storage class for Autoclass configuration.
+        :getter: Get the terminal storage class for Autoclass configuration.
+
+        :rtype: str
+        :returns: The terminal storage class if Autoclass is enabled, else ``None``.
+        """
+        autoclass = self._properties.get("autoclass", {})
+        return autoclass.get("terminalStorageClass", None)
+
+    @autoclass_terminal_storage_class.setter
+    def autoclass_terminal_storage_class(self, value):
+        """The storage class that objects in an Autoclass bucket eventually transition to if
+        they are not read for a certain length of time. Valid values are NEARLINE and ARCHIVE.
+
+        See https://cloud.google.com/storage/docs/using-autoclass for details.
+
+        :type value: str
+        :param value: The only valid values are `"NEARLINE"` and `"ARCHIVE"`.
+        """
+        autoclass = self._properties.get("autoclass", {})
+        autoclass["terminalStorageClass"] = value
+        self._patch_property("autoclass", autoclass)
+
+    @property
+    def autoclass_terminal_storage_class_update_time(self):
+        """The time at which the Autoclass terminal_storage_class field was last updated for this bucket
+        :rtype: datetime.datetime or ``NoneType``
+        :returns: point-in time at which the bucket's terminal_storage_class is last updated, or ``None`` if the property is not set locally.
+        """
+        autoclass = self._properties.get("autoclass")
+        if autoclass is not None:
+            timestamp = autoclass.get("terminalStorageClassUpdateTime")
+            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

@@ -449,23 +449,27 @@ def test_get_set_autoclass(new_bucket_obj, test_bucket, capsys):
     out, _ = capsys.readouterr()
     assert "Autoclass enabled is set to False" in out
     assert bucket.autoclass_toggle_time is None
+    assert bucket.autoclass_terminal_storage_class_update_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
+    assert bucket.autoclass_terminal_storage_class == "NEARLINE"
 
-    # Test disabling Autoclass
-    bucket = storage_set_autoclass.set_autoclass(bucket.name, False)
+    # Test set terminal_storage_class to ARCHIVE
+    bucket = storage_set_autoclass.set_autoclass(bucket.name)
     out, _ = capsys.readouterr()
-    assert "Autoclass enabled is set to False" in out
-    assert bucket.autoclass_enabled is False
+    assert "Autoclass enabled is set to True" in out
+    assert bucket.autoclass_enabled is True
+    assert bucket.autoclass_terminal_storage_class == "ARCHIVE"
 
     # Test get Autoclass
     bucket = storage_get_autoclass.get_autoclass(bucket.name)
     out, _ = capsys.readouterr()
-    assert "Autoclass enabled is set to False" in out
+    assert "Autoclass enabled is set to True" in out
     assert bucket.autoclass_toggle_time is not None
+    assert bucket.autoclass_terminal_storage_class_update_time is not None
 
 
 def test_bucket_lifecycle_management(test_bucket, capsys):

samples/snippets/storage_get_autoclass.py

@@ -29,8 +29,11 @@ def get_autoclass(bucket_name):
     bucket = storage_client.get_bucket(bucket_name)
     autoclass_enabled = bucket.autoclass_enabled
     autoclass_toggle_time = bucket.autoclass_toggle_time
+    terminal_storage_class = bucket.autoclass_terminal_storage_class
+    tsc_update_time = bucket.autoclass_terminal_storage_class_update_time
 
     print(f"Autoclass enabled is set to {autoclass_enabled} for {bucket.name} at {autoclass_toggle_time}.")
+    print(f"Autoclass terminal storage class is set to {terminal_storage_class} for {bucket.name} at {tsc_update_time}.")
 
     return bucket
 

samples/snippets/storage_set_autoclass.py

@@ -20,28 +20,32 @@
 from google.cloud import storage
 
 
-def set_autoclass(bucket_name, toggle):
-    """Disable Autoclass for a bucket.
+def set_autoclass(bucket_name):
+    """Configure the Autoclass setting for a bucket.
 
-    Note: Only patch requests that disable autoclass are currently supported.
-    To enable autoclass, you must set it at bucket creation time.
+    terminal_storage_class field is optional and defaults to NEARLINE if not otherwise specified.
+    Valid terminal_storage_class values are NEARLINE and ARCHIVE.
     """
     # The ID of your GCS bucket
     # bucket_name = "my-bucket"
-    # Boolean toggle - if true, enables Autoclass; if false, disables Autoclass
-    # toggle = False
+    # Enable Autoclass for a bucket. Set enabled to false to disable Autoclass.
+    # Set Autoclass.TerminalStorageClass, valid values are NEARLINE and ARCHIVE.
+    enabled = True
+    terminal_storage_class = "ARCHIVE"
 
     storage_client = storage.Client()
     bucket = storage_client.bucket(bucket_name)
 
-    bucket.autoclass_enabled = toggle
+    bucket.autoclass_enabled = enabled
+    bucket.autoclass_terminal_storage_class = terminal_storage_class
     bucket.patch()
     print(f"Autoclass enabled is set to {bucket.autoclass_enabled} for {bucket.name} at {bucket.autoclass_toggle_time}.")
+    print(f"Autoclass terminal storage class is {bucket.autoclass_terminal_storage_class}.")
 
     return bucket
 
 
 # [END storage_set_autoclass]
 
 if __name__ == "__main__":
-    set_autoclass(bucket_name=sys.argv[1], toggle=sys.argv[2])
+    set_autoclass(bucket_name=sys.argv[1])

tests/system/test_bucket.py

@@ -1047,19 +1047,55 @@ def test_new_bucket_with_autoclass(
     storage_client,
     buckets_to_delete,
 ):
-    # Autoclass can be enabled/disabled via bucket create
+    from google.cloud.storage import constants
+
+    # Autoclass can be enabled 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)
 
+    # Autoclass terminal_storage_class is defaulted to NEARLINE if not specified
     assert bucket.autoclass_enabled is True
+    assert bucket.autoclass_terminal_storage_class == constants.NEARLINE_STORAGE_CLASS
 
     # Autoclass can be enabled/disabled via bucket patch
     bucket.autoclass_enabled = False
     bucket.patch(if_metageneration_match=bucket.metageneration)
 
     assert bucket.autoclass_enabled is False
     assert bucket.autoclass_toggle_time != previous_toggle_time
+
+
+def test_config_autoclass_w_existing_bucket(
+    storage_client,
+    buckets_to_delete,
+):
+    from google.cloud.storage import constants
+
+    bucket_name = _helpers.unique_name("for-autoclass")
+    bucket = storage_client.create_bucket(bucket_name)
+    buckets_to_delete.append(bucket)
+    assert bucket.autoclass_enabled is False
+    assert bucket.autoclass_toggle_time is None
+    assert bucket.autoclass_terminal_storage_class is None
+    assert bucket.autoclass_terminal_storage_class_update_time is None
+
+    # Enable Autoclass on existing buckets with terminal_storage_class set to ARCHIVE
+    bucket.autoclass_enabled = True
+    bucket.autoclass_terminal_storage_class = constants.ARCHIVE_STORAGE_CLASS
+    bucket.patch(if_metageneration_match=bucket.metageneration)
+    previous_tsc_update_time = bucket.autoclass_terminal_storage_class_update_time
+    assert bucket.autoclass_enabled is True
+    assert bucket.autoclass_terminal_storage_class == constants.ARCHIVE_STORAGE_CLASS
+
+    # Configure Autoclass terminal_storage_class to NEARLINE
+    bucket.autoclass_terminal_storage_class = constants.NEARLINE_STORAGE_CLASS
+    bucket.patch(if_metageneration_match=bucket.metageneration)
+    assert bucket.autoclass_enabled is True
+    assert bucket.autoclass_terminal_storage_class == constants.NEARLINE_STORAGE_CLASS
+    assert (
+        bucket.autoclass_terminal_storage_class_update_time != previous_tsc_update_time
+    )

tests/unit/test_bucket.py

@@ -2659,15 +2659,19 @@ def test_autoclass_enabled_getter_and_setter(self):
         self.assertIn("autoclass", bucket._changes)
         self.assertFalse(bucket.autoclass_enabled)
 
-    def test_autoclass_toggle_time_missing(self):
+    def test_autoclass_config_unset(self):
         bucket = self._make_one()
         self.assertIsNone(bucket.autoclass_toggle_time)
+        self.assertIsNone(bucket.autoclass_terminal_storage_class)
+        self.assertIsNone(bucket.autoclass_terminal_storage_class_update_time)
 
         properties = {"autoclass": {}}
         bucket = self._make_one(properties=properties)
         self.assertIsNone(bucket.autoclass_toggle_time)
+        self.assertIsNone(bucket.autoclass_terminal_storage_class)
+        self.assertIsNone(bucket.autoclass_terminal_storage_class_update_time)
 
-    def test_autoclass_toggle_time(self):
+    def test_autoclass_toggle_and_tsc_update_time(self):
         import datetime
         from google.cloud._helpers import _datetime_to_rfc3339
         from google.cloud._helpers import UTC
@@ -2677,10 +2681,31 @@ def test_autoclass_toggle_time(self):
             "autoclass": {
                 "enabled": True,
                 "toggleTime": _datetime_to_rfc3339(effective_time),
+                "terminalStorageClass": "NEARLINE",
+                "terminalStorageClassUpdateTime": _datetime_to_rfc3339(effective_time),
             }
         }
         bucket = self._make_one(properties=properties)
         self.assertEqual(bucket.autoclass_toggle_time, effective_time)
+        self.assertEqual(
+            bucket.autoclass_terminal_storage_class_update_time, effective_time
+        )
+
+    def test_autoclass_tsc_getter_and_setter(self):
+        from google.cloud.storage import constants
+
+        properties = {
+            "autoclass": {"terminalStorageClass": constants.ARCHIVE_STORAGE_CLASS}
+        }
+        bucket = self._make_one(properties=properties)
+        self.assertEqual(
+            bucket.autoclass_terminal_storage_class, constants.ARCHIVE_STORAGE_CLASS
+        )
+        bucket.autoclass_terminal_storage_class = constants.NEARLINE_STORAGE_CLASS
+        self.assertIn("autoclass", bucket._changes)
+        self.assertEqual(
+            bucket.autoclass_terminal_storage_class, constants.NEARLINE_STORAGE_CLASS
+        )
 
     def test_get_logging_w_prefix(self):
         NAME = "name"