Use set instead of list for dags' tags

airflow/models/dag.py

@@ -371,7 +371,7 @@ def _create_orm_dagrun(
  "doc_md": str,
  "is_paused_upon_creation": bool,
  "render_template_as_native_obj": bool,
- "tags": list,
+ "tags": abc.Collection[str],
  "auto_register": bool,
  "fail_stop": bool,
  "dag_display_name": str,
@@ -552,7 +552,7 @@ def __init__(
  is_paused_upon_creation: bool | None = None,
  jinja_environment_kwargs: dict | None = None,
  render_template_as_native_obj: bool = False,
- tags: list[str] | None = None,
+ tags: abc.Collection[str] | None = None,
  owner_links: dict[str, str] | None = None,
  auto_register: bool = True,
  fail_stop: bool = False,
@@ -767,7 +767,7 @@ def __init__(
 
  self.doc_md = self.get_doc_md(doc_md)
 
- self.tags = tags or []
+ self.tags: abc.MutableSet[str] = set(tags or [])
  self._task_group = TaskGroup.create_root(self)
  self.validate_schedule_and_params()
  wrong_links = dict(self.iter_invalid_owner_links())
@@ -2948,7 +2948,7 @@ def bulk_write_to_db(
  else:
  orm_dag.calculate_dagrun_date_fields(dag, last_automated_data_interval)
 
- dag_tags = set(dag.tags or {})
+ dag_tags = dag.tags or set()
  orm_dag_tags = list(orm_dag.tags or [])
  for orm_tag in orm_dag_tags:
  if orm_tag.name not in dag_tags:
@@ -3825,7 +3825,7 @@ def dag(
  is_paused_upon_creation: bool | None = None,
  jinja_environment_kwargs: dict | None = None,
  render_template_as_native_obj: bool = False,
- tags: list[str] | None = None,
+ tags: abc.Collection[str] | None = None,
  owner_links: dict[str, str] | None = None,
  auto_register: bool = True,
  fail_stop: bool = False,

newsfragments/41420.significant.rst

@@ -0,0 +1 @@
+Replaced Python's `list` with `collections.abc.MutableSet` for the `Dag` tags.

tests/models/test_dag.py

@@ -1060,7 +1060,7 @@ def test_bulk_write_to_db(self):
 
  # Removing all tags
  for dag in dags:
- dag.tags = None
+ dag.tags = set()
  with assert_queries_count(9):
  DAG.bulk_write_to_db(dags)
  with create_session() as session:
@@ -3720,6 +3720,30 @@ def test__tags_length(tags: list[str], should_pass: bool):
  DAG("test-dag", schedule=None, tags=tags)
 
 
+@pytest.mark.parametrize(
+ "input_tags, expected_result",
+ [
+ pytest.param([], set(), id="empty tags"),
+ pytest.param(["a normal tag"], {"a normal tag",}, id="one tag"),
+ pytest.param(["a normal tag", "another normal tag"], {"a normal tag", "another normal tag"}, id="two different tags"),
+ pytest.param(["a", "a"], {"a",}, id="two same tags"),
+ ],
+)
+def test__tags_duplicates(input_tags: list[str], expected_result: set[str]):
+ result = DAG("test-dag", tags=input_tags)
+ assert result.tags == expected_result
+
+
+def test__tags_mutable():
+ expected_tags = {"6", "7"}
+ test_dag = DAG("test-dag")
+ test_dag.tags.add("6")
+ test_dag.tags.add("7")
+ test_dag.tags.add("8")
+ test_dag.tags.remove("8")
+ assert test_dag.tags == expected_tags
+
+
 @pytest.mark.need_serialized_dag
 def test_get_dataset_triggered_next_run_info(dag_maker, clear_datasets):
  dataset1 = Dataset(uri="ds1")