[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)
 

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)
 

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)
 

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)
 

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)
 

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)
 

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)
 

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)