feat: support resource_tags for table

google/cloud/bigquery/table.py

@@ -408,6 +408,7 @@ class Table(_TableBase):
         "require_partition_filter": "requirePartitionFilter",
         "table_constraints": "tableConstraints",
         "max_staleness": "maxStaleness",
+        "resource_tags": "resourceTags",
     }
 
     def __init__(self, table_ref, schema=None) -> None:
@@ -1023,6 +1024,25 @@ def table_constraints(self) -> Optional["TableConstraints"]:
             table_constraints = TableConstraints.from_api_repr(table_constraints)
         return table_constraints
 
+    @property
+    def resource_tags(self):
+        """Dict[str, str]: Resource tags for the table.
+
+        See: https://cloud.google.com/bigquery/docs/reference/rest/v2/tables#Table.FIELDS.resource_tags
+
+        Raises:
+            ValueError: For invalid value types.
+        """
+        return self._properties.setdefault(
+            self._PROPERTY_TO_API_FIELD["resource_tags"], {}
+        )
+
+    @resource_tags.setter
+    def resource_tags(self, value):
+        if not isinstance(value, dict) and value is not None:
+            raise ValueError("resource_tags must be a dict or None")
+        self._properties[self._PROPERTY_TO_API_FIELD["resource_tags"]] = value
+
     @classmethod
     def from_string(cls, full_table_id: str) -> "Table":
         """Construct a table from fully-qualified table ID.

noxfile.py

@@ -219,6 +219,9 @@ def system(session):
     # Data Catalog needed for the column ACL test with a real Policy Tag.
     session.install("google-cloud-datacatalog", "-c", constraints_path)
 
+    # Resource Manager needed for test with a real Resource Tag.
+    session.install("google-cloud-resource-manager", "-c", constraints_path)
+
     if session.python in ["3.11", "3.12"]:
         extras = "[bqstorage,ipywidgets,pandas,tqdm,opentelemetry]"
     else:
@@ -366,6 +369,7 @@ def prerelease_deps(session):
     session.install(
         "freezegun",
         "google-cloud-datacatalog",
+        "google-cloud-resource-manager",
         "google-cloud-storage",
         "google-cloud-testutils",
         "psutil",

tests/system/test_client.py

@@ -45,6 +45,8 @@
 from google.cloud import storage
 from google.cloud.datacatalog_v1 import types as datacatalog_types
 from google.cloud.datacatalog_v1 import PolicyTagManagerClient
+from google.cloud.resourcemanager_v3 import types as resourcemanager_types
+from google.cloud.resourcemanager_v3 import TagKeysClient, TagValuesClient
 import psutil
 import pytest
 from test_utils.retry import RetryErrors
@@ -159,6 +161,8 @@ def setUp(self):
 
     def tearDown(self):
         policy_tag_client = PolicyTagManagerClient()
+        tag_keys_client = TagKeysClient()
+        tag_values_client = TagValuesClient()
 
         def _still_in_use(bad_request):
             return any(
@@ -178,6 +182,10 @@ def _still_in_use(bad_request):
                 retry_in_use(Config.CLIENT.delete_table)(doomed)
             elif isinstance(doomed, datacatalog_types.Taxonomy):
                 policy_tag_client.delete_taxonomy(name=doomed.name)
+            elif isinstance(doomed, resourcemanager_types.TagKey):
+                tag_keys_client.delete_tag_key(name=doomed.name).result()
+            elif isinstance(doomed, resourcemanager_types.TagValue):
+                tag_values_client.delete_tag_value(name=doomed.name).result()
             else:
                 doomed.delete()
 
@@ -646,6 +654,30 @@ def test_list_tables(self):
     def test_update_table(self):
         dataset = self.temp_dataset(_make_dataset_id("update_table"))
 
+        def _create_resource_tag_key_and_values(key, values):
+            tag_key_client = TagKeysClient()
+            tag_value_client = TagValuesClient()
+
+            tag_key_parent = f"projects/{Config.CLIENT.project}"
+            new_tag_key = resourcemanager_types.TagKey(
+                short_name=key, parent=tag_key_parent
+            )
+            tag_key = tag_key_client.create_tag_key(tag_key=new_tag_key).result()
+            self.to_delete.insert(0, tag_key)
+
+            for value in values:
+                new_tag_value = resourcemanager_types.TagValue(
+                    short_name=value, parent=tag_key.name
+                )
+                tag_value = tag_value_client.create_tag_value(
+                    tag_value=new_tag_value
+                ).result()
+                self.to_delete.insert(0, tag_value)
+
+        _create_resource_tag_key_and_values("owner", ["Alice", "Bob"])
+        _create_resource_tag_key_and_values("classification", ["public"])
+        _create_resource_tag_key_and_values("env", ["dev"])
+
         TABLE_NAME = "test_table"
         table_arg = Table(dataset.table(TABLE_NAME), schema=SCHEMA)
         self.assertFalse(_table_exists(table_arg))
@@ -658,24 +690,54 @@ def test_update_table(self):
         table.friendly_name = "Friendly"
         table.description = "Description"
         table.labels = {"priority": "high", "color": "blue"}
+        table.resource_tags = {
+            f"{Config.CLIENT.project}/owner": "Alice",
+            f"{Config.CLIENT.project}/env": "dev",
+        }
 
         table2 = Config.CLIENT.update_table(
-            table, ["friendly_name", "description", "labels"]
+            table, ["friendly_name", "description", "labels", "resource_tags"]
         )
 
         self.assertEqual(table2.friendly_name, "Friendly")
         self.assertEqual(table2.description, "Description")
         self.assertEqual(table2.labels, {"priority": "high", "color": "blue"})
+        self.assertEqual(
+            table2.resource_tags,
+            {
+                f"{Config.CLIENT.project}/owner": "Alice",
+                f"{Config.CLIENT.project}/env": "dev",
+            },
+        )
 
         table2.description = None
         table2.labels = {
             "color": "green",  # change
             "shape": "circle",  # add
             "priority": None,  # delete
         }
-        table3 = Config.CLIENT.update_table(table2, ["description", "labels"])
+        table2.resource_tags = {
+            f"{Config.CLIENT.project}/owner": "Bob",  # change
+            f"{Config.CLIENT.project}/classification": "public",  # add
+            f"{Config.CLIENT.project}/env": None,  # delete
+        }
+        table3 = Config.CLIENT.update_table(
+            table2, ["description", "labels", "resource_tags"]
+        )
         self.assertIsNone(table3.description)
         self.assertEqual(table3.labels, {"color": "green", "shape": "circle"})
+        self.assertEqual(
+            table3.resource_tags,
+            {
+                f"{Config.CLIENT.project}/owner": "Bob",
+                f"{Config.CLIENT.project}/classification": "public",
+            },
+        )
+
+        # Delete resource tag bindings.
+        table3.resource_tags = None
+        table4 = Config.CLIENT.update_table(table3, ["resource_tags"])
+        self.assertEqual(table4.resource_tags, {})
 
         # If we try to update using table2 again, it will fail because the
         # previous update changed the ETag.

tests/unit/test_client.py

@@ -2314,6 +2314,7 @@ def test_update_table(self):
                 "description": description,
                 "friendlyName": title,
                 "labels": {"x": "y"},
+                "resourceTags": {"123456789012/key": "value"},
             }
         )
         schema = [
@@ -2337,7 +2338,8 @@ def test_update_table(self):
         table.description = description
         table.friendly_name = title
         table.labels = {"x": "y"}
-        fields = ["schema", "description", "friendly_name", "labels"]
+        table.resource_tags = {"123456789012/key": "value"}
+        fields = ["schema", "description", "friendly_name", "labels", "resource_tags"]
         with mock.patch(
             "google.cloud.bigquery.opentelemetry_tracing._get_final_span_attributes"
         ) as final_attributes:
@@ -2369,14 +2371,16 @@ def test_update_table(self):
             "description": description,
             "friendlyName": title,
             "labels": {"x": "y"},
+            "resourceTags": {"123456789012/key": "value"},
         }
         conn.api_request.assert_called_once_with(
-            method="PATCH", data=sent, path="/" + path, timeout=7.5
+            method="PATCH", path="/" + path, data=sent, timeout=7.5
         )
         self.assertEqual(updated_table.description, table.description)
         self.assertEqual(updated_table.friendly_name, table.friendly_name)
         self.assertEqual(updated_table.schema, table.schema)
         self.assertEqual(updated_table.labels, table.labels)
+        self.assertEqual(updated_table.resource_tags, table.resource_tags)
 
         # ETag becomes If-Match header.
         table._properties["etag"] = "etag"

tests/unit/test_table.py

@@ -1458,6 +1458,33 @@ def test_encryption_configuration_setter(self):
         table.encryption_configuration = None
         self.assertIsNone(table.encryption_configuration)
 
+    def test_resource_tags_getter_empty(self):
+        dataset = DatasetReference(self.PROJECT, self.DS_ID)
+        table_ref = dataset.table(self.TABLE_NAME)
+        table = self._make_one(table_ref)
+        self.assertEqual(table.resource_tags, {})
+
+    def test_resource_tags_update_in_place(self):
+        dataset = DatasetReference(self.PROJECT, self.DS_ID)
+        table_ref = dataset.table(self.TABLE_NAME)
+        table = self._make_one(table_ref)
+        table.resource_tags["123456789012/key"] = "value"
+        self.assertEqual(table.resource_tags, {"123456789012/key": "value"})
+
+    def test_resource_tags_setter(self):
+        dataset = DatasetReference(self.PROJECT, self.DS_ID)
+        table_ref = dataset.table(self.TABLE_NAME)
+        table = self._make_one(table_ref)
+        table.resource_tags = {"123456789012/key": "value"}
+        self.assertEqual(table.resource_tags, {"123456789012/key": "value"})
+
+    def test_resource_tags_setter_bad_value(self):
+        dataset = DatasetReference(self.PROJECT, self.DS_ID)
+        table_ref = dataset.table(self.TABLE_NAME)
+        table = self._make_one(table_ref)
+        with self.assertRaises(ValueError):
+            table.resource_tags = 12345
+
     def test___repr__(self):
         from google.cloud.bigquery.table import TableReference