Use set instead of list for dags' tags

airflow/models/dag.py

@@ -355,7 +355,7 @@ def _create_orm_dagrun(
     "doc_md": str,
     "is_paused_upon_creation": bool,
     "render_template_as_native_obj": bool,
-    "tags": list,
+    "tags": abc.Collection,
     "auto_register": bool,
     "fail_stop": bool,
     "dag_display_name": str,
@@ -532,7 +532,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 | None = None,
         owner_links: dict[str, str] | None = None,
         auto_register: bool = True,
         fail_stop: bool = False,
@@ -682,7 +682,7 @@ def __init__(
 
         self.doc_md = self.get_doc_md(doc_md)
 
-        self.tags = tags or []
+        self.tags: abc.MutableSet = set(tags or [])
         self._task_group = TaskGroup.create_root(self)
         self.validate_schedule_and_params()
         wrong_links = dict(self.iter_invalid_owner_links())
@@ -2741,7 +2741,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
             orm_dag_tags = list(orm_dag.tags or [])
             for orm_tag in orm_dag_tags:
                 if orm_tag.name not in dag_tags:
@@ -3603,7 +3603,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 | None = None,
     owner_links: dict[str, str] | None = None,
     auto_register: bool = True,
     fail_stop: bool = False,

newsfragments/41420.significant.rst

@@ -0,0 +1,11 @@
+**Breaking Change**
+
+Replaced Python's `list` with `collections.abc.MutableSet` for the property `Dag.tags`.
+
+At the constractur you still can use list,
+you actually can use any data structure that implements the
+`collections.abc.Collection` interface.
+
+The `tags` property of the `Dag` model would be of type
+`collections.abc.MutableSet` instead of `list`,
+as there are no actual duplicates at the tags.

tests/models/test_dag.py

@@ -826,7 +826,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:
@@ -3347,6 +3347,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")