Repository: googleapis/python-bigquery
Pull request #2163 (2025-04-16 19:49:26)

feat: Update the AccessEntry class with a new condition attribute and unit tests
Updates the `AccessEntry` class with a new `condition` attribute and unit tests.

Specifically:

* Adds a condition attribute to the AccessEntry class
* Updates the pre-existing Condition class with several dunder methods (__eq__, __hash__, etc)
* Adds unit tests to put the condition setters, getters, etc through their paces
* Adds unit tests to exercise the new Condition dunder methods

This work is in support of internal bug: b/330869964



Patch:
diff --git a/google/cloud/bigquery/dataset.py b/google/cloud/bigquery/dataset.py
index cc14598f..670fe127 100644
--- a/google/cloud/bigquery/dataset.py
+++ b/google/cloud/bigquery/dataset.py
@@ -298,12 +298,15 @@ class AccessEntry(object):
         role: Optional[str] = None,
         entity_type: Optional[str] = None,
         entity_id: Optional[Union[Dict[str, Any], str]] = None,
+        **kwargs,
     ):
-        self._properties = {}
+        self._properties: Dict[str, Any] = {}
         if entity_type is not None:
             self._properties[entity_type] = entity_id
         self._properties["role"] = role
-        self._entity_type = entity_type
+        self._entity_type: Optional[str] = entity_type
+        for prop, val in kwargs.items():
+            setattr(self, prop, val)
 
     @property
     def role(self) -> Optional[str]:
@@ -330,6 +333,9 @@ class AccessEntry(object):
         if isinstance(value, str):
             value = DatasetReference.from_string(value).to_api_repr()
 
+        if isinstance(value, DatasetReference):
+            value = value.to_api_repr()
+
         if isinstance(value, (Dataset, DatasetListItem)):
             value = value.reference.to_api_repr()
 
@@ -437,15 +443,65 @@ class AccessEntry(object):
     def special_group(self, value):
         self._properties["specialGroup"] = value
 
+    @property
+    def condition(self) -> Optional["Condition"]:
+        """Optional[Condition]: The IAM condition associated with this entry."""
+        value = typing.cast(Dict[str, Any], self._properties.get("condition"))
+        return Condition.from_api_repr(value) if value else None
+
+    @condition.setter
+    def condition(self, value: Union["Condition", dict, None]):
+        """Set the IAM condition for this entry."""
+        if value is None:
+            self._properties["condition"] = None
+        elif isinstance(value, Condition):
+            self._properties["condition"] = value.to_api_repr()
+        elif isinstance(value, dict):
+            self._properties["condition"] = value
+        else:
+            raise TypeError("condition must be a Condition object, dict, or None")
+
     @property
     def entity_type(self) -> Optional[str]:
         """The entity_type of the entry."""
+
+        # The api_repr for an AccessEntry object is expected to be a dict with
+        # only a few keys. Two keys that may be present are role and condition.
+        # Any additional key is going to have one of ~eight different names:
+        #   userByEmail, groupByEmail, domain, dataset, specialGroup, view,
+        #   routine, iamMember
+
+        # if self._entity_type is None, see if it needs setting
+        # i.e. is there a key: value pair that should be associated with
+        # entity_type and entity_id?
+        if self._entity_type is None:
+            resource = self._properties.copy()
+            # we are empyting the dict to get to the last `key: value`` pair
+            # so we don't keep these first entries
+            _ = resource.pop("role", None)
+            _ = resource.pop("condition", None)
+
+            try:
+                # we only need entity_type, because entity_id gets set elsewhere.
+                entity_type, _ = resource.popitem()
+            except KeyError:
+                entity_type = None
+
+            self._entity_type = entity_type
+
         return self._entity_type
 
     @property
     def entity_id(self) -> Optional[Union[Dict[str, Any], str]]:
         """The entity_id of the entry."""
-        return self._properties.get(self._entity_type) if self._entity_type else None
+        if self.entity_type:
+            entity_type = self.entity_type
+        else:
+            return None
+        return typing.cast(
+            Optional[Union[Dict[str, Any], str]],
+            self._properties.get(entity_type, None),
+        )
 
     def __eq__(self, other):
         if not isinstance(other, AccessEntry):
@@ -464,7 +520,16 @@ class AccessEntry(object):
         Returns:
             Tuple: The contents of this :class:`~google.cloud.bigquery.dataset.AccessEntry`.
         """
+
         properties = self._properties.copy()
+
+        # Dicts are not hashable.
+        # Convert condition to a hashable datatype(s)
+        condition = properties.get("condition")
+        if isinstance(condition, dict):
+            condition_key = tuple(sorted(condition.items()))
+            properties["condition"] = condition_key
+
         prop_tup = tuple(sorted(properties.items()))
         return (self.role, self._entity_type, self.entity_id, prop_tup)
 
@@ -491,19 +556,11 @@ class AccessEntry(object):
         Returns:
             google.cloud.bigquery.dataset.AccessEntry:
                 Access entry parsed from ``resource``.
-
-        Raises:
-            ValueError:
-                If the resource has more keys than ``role`` and one additional
-                key.
         """
-        entry = resource.copy()
-        role = entry.pop("role", None)
-        entity_type, entity_id = entry.popitem()
-        if len(entry) != 0:
-            raise ValueError("Entry has unexpected keys remaining.", entry)
 
-        return cls(role, entity_type, entity_id)
+        access_entry = cls()
+        access_entry._properties = resource.copy()
+        return access_entry
 
 
 class Dataset(object):
@@ -1160,6 +1217,43 @@ class Condition(object):
 
         return cls(
             expression=resource["expression"],
-            title=resource.get("title"),
-            description=resource.get("description"),
+            title=resource.get("title", None),
+            description=resource.get("description", None),
         )
+
+    def __eq__(self, other: object) -> bool:
+        """Check for equality based on expression, title, and description."""
+        if not isinstance(other, Condition):
+            return NotImplemented
+        return self._key() == other._key()
+
+    def _key(self):
+        """A tuple key that uniquely describes this field.
+        Used to compute this instance's hashcode and evaluate equality.
+        Returns:
+            Tuple: The contents of this :class:`~google.cloud.bigquery.dataset.AccessEntry`.
+        """
+
+        properties = self._properties.copy()
+
+        # Dicts are not hashable.
+        # Convert object to a hashable datatype(s)
+        prop_tup = tuple(sorted(properties.items()))
+        return prop_tup
+
+    def __ne__(self, other: object) -> bool:
+        """Check for inequality."""
+        return not self == other
+
+    def __hash__(self) -> int:
+        """Generate a hash based on expression, title, and description."""
+        return hash(self._key())
+
+    def __repr__(self) -> str:
+        """Return a string representation of the Condition object."""
+        parts = [f"expression={self.expression!r}"]
+        if self.title is not None:
+            parts.append(f"title={self.title!r}")
+        if self.description is not None:
+            parts.append(f"description={self.description!r}")
+        return f"Condition({', '.join(parts)})"


Test patch:
diff --git a/tests/unit/test_dataset.py b/tests/unit/test_dataset.py
index 036e2245..51f1809b 100644
--- a/tests/unit/test_dataset.py
+++ b/tests/unit/test_dataset.py
@@ -167,7 +167,10 @@ class TestAccessEntry(unittest.TestCase):
             entity_type="view",
             entity_id=resource["view"],
         )
-        self.assertEqual(entry, exp_entry)
+
+        assert entry.entity_type == exp_entry.entity_type
+        assert entry.entity_id == exp_entry.entity_id
+        assert entry.role is None
 
     def test_to_api_repr_w_extra_properties(self):
         resource = {
@@ -179,15 +182,6 @@ class TestAccessEntry(unittest.TestCase):
         exp_resource = entry.to_api_repr()
         self.assertEqual(resource, exp_resource)
 
-    def test_from_api_repr_entries_w_extra_keys(self):
-        resource = {
-            "role": "READER",
-            "specialGroup": "projectReaders",
-            "userByEmail": "salmon@example.com",
-        }
-        with self.assertRaises(ValueError):
-            self._get_target_class().from_api_repr(resource)
-
     def test_view_getter_setter(self):
         view = {
             "projectId": "my_project",
@@ -307,7 +301,10 @@ class TestAccessEntry(unittest.TestCase):
         entry.dataset = dataset_ref
         resource = entry.to_api_repr()
         exp_resource = {
-            "dataset": {"dataset": dataset_ref, "targetTypes": None},
+            "dataset": {
+                "dataset": {"datasetId": "my_dataset", "projectId": "my-project"},
+                "targetTypes": None,
+            },
             "role": None,
         }
         self.assertEqual(resource, exp_resource)
@@ -494,6 +491,262 @@ class TestAccessEntry(unittest.TestCase):
         self.assertEqual(entry.dataset_target_types, target_types)
 
 
+# --- Tests for AccessEntry when using Condition ---
+
+EXPRESSION = "request.time < timestamp('2026-01-01T00:00:00Z')"
+TITLE = "Expires end 2025"
+DESCRIPTION = "Access expires at the start of 2026."
+
+
+@pytest.fixture
+def condition_1():
+    """Provides a sample Condition object."""
+    return Condition(
+        expression=EXPRESSION,
+        title=TITLE,
+        description=DESCRIPTION,
+    )
+
+
+@pytest.fixture
+def condition_1_api_repr():
+    """Provides the API representation for condition_1."""
+    # Use the actual to_api_repr method
+    return Condition(
+        expression=EXPRESSION,
+        title=TITLE,
+        description=DESCRIPTION,
+    ).to_api_repr()
+
+
+@pytest.fixture
+def condition_2():
+    """Provides a second, different Condition object."""
+    return Condition(
+        expression="resource.name.startsWith('projects/_/buckets/restricted/')",
+        title="Restricted Buckets",
+    )
+
+
+@pytest.fixture
+def condition_2_api_repr():
+    """Provides the API representation for condition2."""
+    # Use the actual to_api_repr method
+    return Condition(
+        expression="resource.name.startsWith('projects/_/buckets/restricted/')",
+        title="Restricted Buckets",
+    ).to_api_repr()
+
+
+class TestAccessEntryAndCondition:
+    @staticmethod
+    def _get_target_class():
+        return AccessEntry
+
+    def _make_one(self, *args, **kw):
+        return self._get_target_class()(*args, **kw)
+
+    # Test __init__ without condition
+    def test_init_without_condition(self):
+        entry = AccessEntry("READER", "userByEmail", "test@example.com")
+        assert entry.role == "READER"
+        assert entry.entity_type == "userByEmail"
+        assert entry.entity_id == "test@example.com"
+        assert entry.condition is None
+        # Accessing _properties is for internal verification in tests
+        assert "condition" not in entry._properties
+
+    # Test __init__ with condition object
+    def test_init_with_condition_object(self, condition_1, condition_1_api_repr):
+        entry = AccessEntry(
+            "READER", "userByEmail", "test@example.com", condition=condition_1
+        )
+        assert entry.condition == condition_1
+        assert entry._properties.get("condition") == condition_1_api_repr
+
+    # Test __init__ with condition=None
+    def test_init_with_condition_none(self):
+        entry = AccessEntry("READER", "userByEmail", "test@example.com", condition=None)
+        assert entry.condition is None
+
+    # Test condition getter/setter
+    def test_condition_getter_setter(
+        self, condition_1, condition_1_api_repr, condition_2, condition_2_api_repr
+    ):
+        entry = AccessEntry("WRITER", "group", "admins@example.com")
+        assert entry.condition is None
+
+        # Set condition 1
+        entry.condition = condition_1
+        assert entry.condition.to_api_repr() == condition_1_api_repr
+        assert entry._properties.get("condition") == condition_1_api_repr
+
+        # Set condition 2
+        entry.condition = condition_2
+        assert entry.condition.to_api_repr() == condition_2_api_repr
+        assert entry._properties.get("condition") != condition_1_api_repr
+        assert entry._properties.get("condition") == condition_2.to_api_repr()
+
+        # Set back to None
+        entry.condition = None
+        assert entry.condition is None
+
+        # Set condition using a dict
+        entry.condition = condition_1_api_repr
+        assert entry._properties.get("condition") == condition_1_api_repr
+
+    # Test setter validation
+    def test_condition_setter_invalid_type(self):
+        entry = AccessEntry("READER", "domain", "example.com")
+        with pytest.raises(
+            TypeError, match="condition must be a Condition object, dict, or None"
+        ):
+            entry.condition = 123  # type: ignore
+
+    # Test equality/hash without condition
+    def test_equality_and_hash_without_condition(self):
+        entry1 = AccessEntry("OWNER", "specialGroup", "projectOwners")
+        entry2 = AccessEntry("OWNER", "specialGroup", "projectOwners")
+        entry3 = AccessEntry("WRITER", "specialGroup", "projectOwners")
+        assert entry1 == entry2
+        assert entry1 != entry3
+        assert hash(entry1) == hash(entry2)
+        assert hash(entry1) != hash(entry3)  # Usually true
+
+    def test_equality_and_hash_with_condition(self, condition_1, condition_2):
+        cond1a = Condition(
+            condition_1.expression, condition_1.title, condition_1.description
+        )
+        cond1b = Condition(
+            condition_1.expression, condition_1.title, condition_1.description
+        )  # Same values, different object
+
+        entry1a = AccessEntry(
+            "READER", "userByEmail", "a@example.com", condition=cond1a
+        )
+        entry1b = AccessEntry(
+            "READER", "userByEmail", "a@example.com", condition=cond1b
+        )  # Different Condition instance
+        entry2 = AccessEntry(
+            "READER", "userByEmail", "a@example.com", condition=condition_2
+        )
+        entry3 = AccessEntry("READER", "userByEmail", "a@example.com")  # No condition
+        entry4 = AccessEntry(
+            "WRITER", "userByEmail", "a@example.com", condition=cond1a
+        )  # Different role
+
+        assert entry1a == entry1b
+        assert entry1a != entry2
+        assert entry1a != entry3
+        assert entry1a != entry4
+        assert entry2 != entry3
+
+        assert hash(entry1a) == hash(entry1b)
+        assert hash(entry1a) != hash(entry2)  # Usually true
+        assert hash(entry1a) != hash(entry3)  # Usually true
+        assert hash(entry1a) != hash(entry4)  # Usually true
+
+    # Test to_api_repr with condition
+    def test_to_api_repr_with_condition(self, condition_1, condition_1_api_repr):
+        entry = AccessEntry(
+            "WRITER", "groupByEmail", "editors@example.com", condition=condition_1
+        )
+        expected_repr = {
+            "role": "WRITER",
+            "groupByEmail": "editors@example.com",
+            "condition": condition_1_api_repr,
+        }
+        assert entry.to_api_repr() == expected_repr
+
+    def test_view_property_with_condition(self, condition_1):
+        """Test setting/getting view property when condition is present."""
+        entry = AccessEntry(role=None, entity_type="view", condition=condition_1)
+        view_ref = TableReference(DatasetReference("proj", "dset"), "view_tbl")
+        entry.view = view_ref  # Use the setter
+        assert entry.view == view_ref
+        assert entry.condition == condition_1  # Condition should persist
+        assert entry.role is None
+        assert entry.entity_type == "view"
+
+        # Check internal representation
+        assert "view" in entry._properties
+        assert "condition" in entry._properties
+
+    def test_user_by_email_property_with_condition(self, condition_1):
+        """Test setting/getting user_by_email property when condition is present."""
+        entry = AccessEntry(
+            role="READER", entity_type="userByEmail", condition=condition_1
+        )
+        email = "test@example.com"
+        entry.user_by_email = email  # Use the setter
+        assert entry.user_by_email == email
+        assert entry.condition == condition_1  # Condition should persist
+        assert entry.role == "READER"
+        assert entry.entity_type == "userByEmail"
+
+        # Check internal representation
+        assert "userByEmail" in entry._properties
+        assert "condition" in entry._properties
+
+    # Test from_api_repr without condition
+    def test_from_api_repr_without_condition(self):
+        api_repr = {"role": "OWNER", "userByEmail": "owner@example.com"}
+        entry = AccessEntry.from_api_repr(api_repr)
+        assert entry.role == "OWNER"
+        assert entry.entity_type == "userByEmail"
+        assert entry.entity_id == "owner@example.com"
+        assert entry.condition is None
+
+    # Test from_api_repr with condition
+    def test_from_api_repr_with_condition(self, condition_1, condition_1_api_repr):
+        api_repr = {
+            "role": "READER",
+            "view": {"projectId": "p", "datasetId": "d", "tableId": "v"},
+            "condition": condition_1_api_repr,
+        }
+        entry = AccessEntry.from_api_repr(api_repr)
+        assert entry.role == "READER"
+        assert entry.entity_type == "view"
+        # The entity_id for view/routine/dataset is the dict itself
+        assert entry.entity_id == {"projectId": "p", "datasetId": "d", "tableId": "v"}
+        assert entry.condition == condition_1
+
+    # Test from_api_repr edge case
+    def test_from_api_repr_no_entity(self, condition_1, condition_1_api_repr):
+        api_repr = {"role": "READER", "condition": condition_1_api_repr}
+        entry = AccessEntry.from_api_repr(api_repr)
+        assert entry.role == "READER"
+        assert entry.entity_type is None
+        assert entry.entity_id is None
+        assert entry.condition == condition_1
+
+    def test_dataset_property_with_condition(self, condition_1):
+        project = "my-project"
+        dataset_id = "my_dataset"
+        dataset_ref = DatasetReference(project, dataset_id)
+        entry = self._make_one(None)
+        entry.dataset = dataset_ref
+        entry.condition = condition_1
+
+        resource = entry.to_api_repr()
+        exp_resource = {
+            "role": None,
+            "dataset": {
+                "dataset": {"datasetId": "my_dataset", "projectId": "my-project"},
+                "targetTypes": None,
+            },
+            "condition": {
+                "expression": "request.time < timestamp('2026-01-01T00:00:00Z')",
+                "title": "Expires end 2025",
+                "description": "Access expires at the start of 2026.",
+            },
+        }
+        assert resource == exp_resource
+        # Check internal representation
+        assert "dataset" in entry._properties
+        assert "condition" in entry._properties
+
+
 class TestDatasetReference(unittest.TestCase):
     @staticmethod
     def _get_target_class():
@@ -821,7 +1074,15 @@ class TestDataset(unittest.TestCase):
         self.assertEqual(
             dataset.path, "/projects/%s/datasets/%s" % (OTHER_PROJECT, self.DS_ID)
         )
-        self.assertEqual(dataset.access_entries, entries)
+        # creating a list of entries relies on AccessEntry.from_api_repr
+        # which does not create an object in exactly the same way as calling the
+        # class directly. We rely on calls to .entity_type and .entity_id to
+        # finalize the settings on each class.
+        entry_pairs = zip(dataset.access_entries, entries)
+        for pair in entry_pairs:
+            assert pair[0].role == pair[1].role
+            assert pair[0].entity_type == pair[1].entity_type
+            assert pair[0].entity_id == pair[1].entity_id
 
         self.assertIsNone(dataset.created)
         self.assertIsNone(dataset.full_dataset_id)
@@ -854,8 +1115,18 @@ class TestDataset(unittest.TestCase):
         dataset = self._make_one(self.DS_REF)
         phred = AccessEntry("OWNER", "userByEmail", "phred@example.com")
         bharney = AccessEntry("OWNER", "userByEmail", "bharney@example.com")
-        dataset.access_entries = [phred, bharney]
-        self.assertEqual(dataset.access_entries, [phred, bharney])
+        entries = [phred, bharney]
+        dataset.access_entries = entries
+
+        # creating a list of entries relies on AccessEntry.from_api_repr
+        # which does not create an object in exactly the same way as calling the
+        # class directly. We rely on calls to .entity_type and .entity_id to
+        # finalize the settings on each class.
+        entry_pairs = zip(dataset.access_entries, entries)
+        for pair in entry_pairs:
+            assert pair[0].role == pair[1].role
+            assert pair[0].entity_type == pair[1].entity_type
+            assert pair[0].entity_id == pair[1].entity_id
 
     def test_default_partition_expiration_ms(self):
         dataset = self._make_one("proj.dset")
@@ -1383,3 +1654,40 @@ class TestCondition:
             ValueError, match="API representation missing required 'expression' field."
         ):
             Condition.from_api_repr(api_repr)
+
+    def test___eq___equality(self, condition_1):
+        result = condition_1
+        expected = condition_1
+        assert result == expected
+
+    def test___eq___equality_not_condition(self, condition_1):
+        result = condition_1
+        other = "not a condition"
+        expected = result.__eq__(other)
+        assert expected is NotImplemented
+
+    def test__ne__not_equality(self):
+        result = condition_1
+        expected = condition_2
+        assert result != expected
+
+    def test__hash__function(self, condition_2):
+        cond1 = Condition(
+            expression=self.EXPRESSION, title=self.TITLE, description=self.DESCRIPTION
+        )
+        cond2 = cond1
+        cond_not_equal = condition_2
+        assert cond1 == cond2
+        assert cond1 is cond2
+        assert hash(cond1) == hash(cond2)
+        assert hash(cond1) is not None
+        assert cond_not_equal != cond1
+        assert hash(cond_not_equal) != hash(cond1)
+
+    def test__hash__with_minimal_inputs(self):
+        cond1 = Condition(
+            expression="example",
+            title=None,
+            description=None,
+        )
+        assert hash(cond1) is not None