[v3-1-test] Enable ruff PLW1641 rule (#57679)

airflow-core/src/airflow/serialization/serialized_objects.py

@@ -1345,6 +1345,9 @@ def __eq__(self, other: Any) -> bool:
             getattr(self, c, None) == getattr(other, c, None) for c in BaseOperator._comps
         )
 
+    def __hash__(self):
+        return hash((self.task_type, *[getattr(self, c, None) for c in BaseOperator._comps]))
+
     def __repr__(self) -> str:
         return f"<SerializedTask({self.task_type}): {self.task_id}>"
 

airflow-core/src/airflow/task/priority_strategy.py

@@ -73,6 +73,9 @@ def __eq__(self, other: object) -> bool:
             return False
         return self.serialize() == other.serialize()
 
+    def __hash__(self):
+        return hash(self.serialize())
+
 
 class _AbsolutePriorityWeightStrategy(PriorityWeightStrategy):
     """Priority weight strategy that uses the task's priority weight directly."""

airflow-core/src/airflow/timetables/_cron.py

@@ -128,6 +128,9 @@ def __eq__(self, other: object) -> bool:
             return NotImplemented
         return self._expression == other._expression and self._timezone == other._timezone
 
+    def __hash__(self):
+        return hash((self._expression, self._timezone))
+
     @property
     def summary(self) -> str:
         return self._expression

airflow-core/src/airflow/timetables/interval.py

@@ -201,6 +201,9 @@ def __eq__(self, other: object) -> bool:
             return NotImplemented
         return self._delta == other._delta
 
+    def __hash__(self):
+        return hash(self._delta)
+
     def serialize(self) -> dict[str, Any]:
         from airflow.serialization.serialized_objects import encode_relativedelta
 

airflow-core/src/airflow/timetables/simple.py

@@ -50,6 +50,9 @@ def __eq__(self, other: object) -> bool:
             return NotImplemented
         return True
 
+    def __hash__(self):
+        return hash(self.__class__.__name__)
+
     def serialize(self) -> dict[str, Any]:
         return {}
 

airflow-core/src/airflow/utils/db.py

@@ -1596,6 +1596,9 @@ def __eq__(self, other: object) -> bool:
         z = itertools.zip_longest(iter(self), iter(other), fillvalue=object())
         return all(x == y for x, y in z)
 
+    def __hash__(self):
+        return hash(tuple(x for x in iter(self)))
+
     def __reversed__(self) -> Iterator[T]:
         return iter(self._process_row(r) for r in self._session.execute(self._select_desc))
 

airflow-core/tests/unit/models/test_renderedtifields.py

@@ -70,6 +70,9 @@ def __repr__(self):
     def __eq__(self, other):
         return self.__dict__ == other.__dict__
 
+    def __hash__(self):
+        return hash(self.__dict__)
+
     def __ne__(self, other):
         return not self.__eq__(other)
 

airflow-core/tests/unit/serialization/test_dag_serialization.py

@@ -1339,6 +1339,9 @@ def __repr__(self):
         def __eq__(self, other):
             return self.__dict__ == other.__dict__
 
+        def __hash__(self):
+            return hash(self.__dict__)
+
         def __ne__(self, other):
             return not self.__eq__(other)
 

airflow-core/tests/unit/serialization/test_serde.py

@@ -142,6 +142,9 @@ def deserialize(data: dict, version: int):
     def __eq__(self, other):
         return self.x == other.x
 
+    def __hash__(self):
+        return hash(self.x)
+
 
 @attr.define
 class Y:

airflow-core/tests/unit/utils/test_json.py

@@ -45,6 +45,9 @@ def deserialize(data: dict, version: int):
     def __eq__(self, other):
         return self.x == other.x
 
+    def __hash__(self):
+        return hash(self.x)
+
 
 @dataclass
 class U:

airflow-core/tests/unit/utils/test_sqlalchemy.py

@@ -264,7 +264,7 @@ def test_compare_values(self):
         under older kubernetes library version.
         """
 
-        class MockAttrError:
+        class MockAttrError:  # noqa: PLW1641
             def __eq__(self, other):
                 raise AttributeError("hello")
 

dev/airflow_perf/sql_queries.py

@@ -101,6 +101,9 @@ def __eq__(self, other):
             and self.file == other.file
         )
 
+    def __hash__(self):
+        return hash((self.function, self.sql, self.location, self.file))
+
     def to_dict(self):
         """
         Convert selected attributes of the instance into a dictionary.

devel-common/src/sphinx_exts/docs_build/errors.py

@@ -45,6 +45,9 @@ def __eq__(self, other):
         right = (other.file_path, other.line_no, other.message)
         return left == right
 
+    def __hash__(self):
+        return hash((self.file_path, self.line_no, self.message))
+
     def __ne__(self, other):
         return not self == other
 

devel-common/src/sphinx_exts/docs_build/spelling_checks.py

@@ -61,6 +61,9 @@ def __eq__(self, other):
         )
         return left == right
 
+    def __hash__(self):
+        return hash((self.file_path, self.line_no, self.spelling, self.context_line, self.message))
+
     def __ne__(self, other):
         return not self == other
 

devel-common/src/tests_common/test_utils/timetables.py

@@ -45,6 +45,9 @@ def __eq__(self, other) -> bool:
             return False
         return self.value == other.value
 
+    def __hash__(self):
+        return hash(self.value)
+
     def serialize(self):
         return {"value": self.value}
 

kubernetes-tests/tests/kubernetes_tests/test_base.py

@@ -53,6 +53,9 @@ class StringContainingId(str):
     def __eq__(self, other):
         return self in other.strip() or self in other
 
+    def __hash__(self):
+        return hash(self)
+
 
 class BaseK8STest:
     """Base class for K8S Tests."""

providers/celery/tests/integration/celery/test_celery_executor.py

@@ -312,6 +312,9 @@ def __repr__(self):
     def __eq__(self, other):
         return self.__dict__ == other.__dict__
 
+    def __hash__(self):
+        return hash(self.__dict__)
+
     def __ne__(self, other):
         return not self.__eq__(other)
 

providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/secret.py

@@ -121,5 +121,8 @@ def __eq__(self, other):
             and self.key == other.key
         )
 
+    def __hash__(self):
+        return hash((self.deploy_type, self.deploy_target, self.secret, self.key))
+
     def __repr__(self):
         return f"Secret({self.deploy_type}, {self.deploy_target}, {self.secret}, {self.key})"

providers/databricks/src/airflow/providers/databricks/hooks/databricks.py

@@ -134,6 +134,9 @@ def __eq__(self, other: object) -> bool:
             and self.state_message == other.state_message
         )
 
+    def __hash__(self):
+        return hash((self.life_cycle_state, self.result_state, self.state_message))
+
     def __repr__(self) -> str:
         return str(self.__dict__)
 
@@ -183,6 +186,9 @@ def is_running(self) -> bool:
     def __eq__(self, other) -> bool:
         return self.state == other.state and self.state_message == other.state_message
 
+    def __hash__(self):
+        return hash((self.state, self.state_message))
+
     def __repr__(self) -> str:
         return str(self.__dict__)
 
@@ -244,6 +250,9 @@ def __eq__(self, other: object) -> bool:
             and self.error_message == other.error_message
         )
 
+    def __hash__(self):
+        return hash((self.state, self.error_code, self.error_message))
+
     def __repr__(self) -> str:
         return str(self.__dict__)
 

providers/fab/src/airflow/providers/fab/auth_manager/models/__init__.py

@@ -91,6 +91,9 @@ class Resource(Model):
     def __eq__(self, other):
         return (isinstance(other, self.__class__)) and (self.name == other.name)
 
+    def __hash__(self):
+        return hash((self.id, self.name))
+
     def __neq__(self, other):
         return self.name != other.name
 

providers/snowflake/tests/unit/snowflake/operators/test_snowflake_sql.py

@@ -42,6 +42,9 @@ def __init__(self, **kwargs):
     def __eq__(self, other):
         return isinstance(other, MockRow) and self.__dict__ == other.__dict__
 
+    def __hash__(self):
+        return hash(self.__dict__)
+
     def __repr__(self):
         return f"MockRow({self.__dict__})"
 

providers/weaviate/tests/unit/weaviate/hooks/test_weaviate.py

@@ -100,6 +100,9 @@ def __eq__(self, other: object) -> bool:
             return False
         return self.properties == other.properties and self.uuid == other.uuid
 
+    def __hash__(self):
+        return hash((self.properties, self.uuid))
+
 
 class TestWeaviateHook:
     """

pyproject.toml

@@ -605,6 +605,7 @@ extend-select = [
     "PLW1507", # Shallow copy of os.environ via copy.copy(os.environ)
     "PLW1508", # Invalid type for environment variable default; expected str or None
     "PLW1510", # subprocess.run without explicit check argument
+    "PLW1641", # Object does not implement __hash__ method
     # Per rule enables
     "RUF006", # Checks for asyncio dangling task
     "RUF015", # Checks for unnecessary iterable allocation for first element

task-sdk/src/airflow/sdk/definitions/asset/__init__.py

@@ -426,6 +426,10 @@ def __eq__(self, other: Any) -> bool:
         f = attrs.filters.include(*attrs.fields_dict(Asset))
         return attrs.asdict(self, filter=f) == attrs.asdict(other, filter=f)
 
+    def __hash__(self):
+        f = attrs.filters.include(*attrs.fields_dict(Asset))
+        return hash(attrs.asdict(self, filter=f))
+
     @property
     def normalized_uri(self) -> str | None:
         """

task-sdk/src/airflow/sdk/definitions/operator_resources.py

@@ -55,6 +55,9 @@ def __eq__(self, other: object) -> bool:
             return NotImplemented
         return self.__dict__ == other.__dict__
 
+    def __hash__(self):
+        return hash(self.__dict__)
+
     def __repr__(self):
         return str(self.__dict__)
 
@@ -138,6 +141,9 @@ def __eq__(self, other: object) -> bool:
             return NotImplemented
         return self.__dict__ == other.__dict__
 
+    def __hash__(self):
+        return hash(self.__dict__)
+
     def __repr__(self):
         return str(self.__dict__)
 

task-sdk/src/airflow/sdk/definitions/param.py

@@ -159,6 +159,9 @@ def __eq__(self, other: Any) -> bool:
             return self.dump() == other
         return NotImplemented
 
+    def __hash__(self):
+        return hash(self.dump())
+
     def __copy__(self) -> ParamsDict:
         return ParamsDict(self.__dict, self.suppress_exception)
 

task-sdk/src/airflow/sdk/execution_time/context.py

@@ -351,6 +351,9 @@ def __eq__(self, other):
         # All instances of ConnectionAccessor are equal since it is a stateless dynamic accessor
         return True
 
+    def __hash__(self):
+        return hash(self.__class__.__name__)
+
     def get(self, conn_id: str, default_conn: Any = None) -> Any:
         from airflow.exceptions import AirflowNotFoundException
 
@@ -376,6 +379,9 @@ def __eq__(self, other):
         # All instances of VariableAccessor are equal since it is a stateless dynamic accessor
         return True
 
+    def __hash__(self):
+        return hash(self.__class__.__name__)
+
     def __repr__(self) -> str:
         return "<VariableAccessor (dynamic access)>"
 
@@ -412,6 +418,9 @@ def __eq__(self, other: object) -> bool:
             return False
         return True
 
+    def __hash__(self):
+        return hash(self.__class__.__name__)
+
 
 class _AssetRefResolutionMixin:
     _asset_ref_cache: dict[AssetRef, AssetUniqueKey] = {}

task-sdk/src/airflow/sdk/execution_time/lazy_sequence.py

@@ -81,6 +81,9 @@ def __eq__(self, other: Any) -> bool:
         z = itertools.zip_longest(iter(self), iter(other), fillvalue=object())
         return all(x == y for x, y in z)
 
+    def __hash__(self):
+        return hash((*[item for item in iter(self)],))
+
     def __iter__(self) -> Iterator[T]:
         return LazyXComIterator(seq=self)
 

task-sdk/src/airflow/sdk/io/store.py

@@ -120,6 +120,9 @@ def __eq__(self, other: object) -> bool:
         except ValueError:
             return False
 
+    def __hash__(self):
+        return hash((self.conn_id, self.fsid))
+
 
 _STORE_CACHE: dict[str, ObjectStore] = {}
 

task-sdk/tests/task_sdk/bases/test_operator.py

@@ -63,6 +63,9 @@ def __repr__(self):
     def __eq__(self, other):
         return self.__dict__ == other.__dict__
 
+    def __hash__(self):
+        return hash(self.__dict__)
+
     def __ne__(self, other):
         return not self.__eq__(other)