allows users to write dag_id and task_id in their national characters, added display name for dag / task

airflow/migrations/versions/0123_2_5_0_add_display_name_for_dag_and_task_.py

@@ -0,0 +1,46 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+"""add display name for dag and task instance
+
+Revision ID: d83579315023
+Revises: 290244fb8b83
+Create Date: 2022-12-06 20:26:07.521273
+
+"""
+
+from __future__ import annotations
+
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = "d83579315023"
+down_revision = "290244fb8b83"
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    op.add_column("dag", sa.Column("display_name", sa.Text(), nullable=True))
+    op.add_column("task_instance", sa.Column("display_name", sa.Text(), nullable=True))
+
+
+def downgrade():
+    op.drop_column("dag", "display_name")
+    op.drop_column("task_instance", "display_name")

airflow/models/baseoperator.py

@@ -49,6 +49,7 @@
 import attr
 import pendulum
 from dateutil.relativedelta import relativedelta
+from slugify import slugify
 from sqlalchemy.orm import Session
 from sqlalchemy.orm.exc import NoResultFound
 
@@ -750,6 +751,9 @@ def __init__(
                 category=RemovedInAirflow3Warning,
                 stacklevel=3,
             )
+
+        self.display_name = task_id
+        task_id = slugify(task_id, separator="_")
         validate_key(task_id)
 
         dag = dag or DagContext.get_current_dag()

airflow/models/dag.py

@@ -53,6 +53,7 @@
 import pendulum
 from dateutil.relativedelta import relativedelta
 from pendulum.tz.timezone import Timezone
+from slugify import slugify
 from sqlalchemy import Boolean, Column, ForeignKey, Index, Integer, String, Text, and_, case, func, not_, or_
 from sqlalchemy.ext.associationproxy import association_proxy
 from sqlalchemy.orm import backref, joinedload, relationship
@@ -440,9 +441,12 @@ def __init__(
                 stacklevel=2,
             )
 
-        validate_key(dag_id)
+        dag_id_slugified = slugify(dag_id, separator="_")
+        validate_key(dag_id_slugified)
+
+        self._dag_id = dag_id_slugified
+        self._display_name = dag_id
 
-        self._dag_id = dag_id
         if concurrency:
             # TODO: Remove in Airflow 3.0
             warnings.warn(
@@ -1155,6 +1159,10 @@ def access_control(self):
     def access_control(self, value):
         self._access_control = DAG._upgrade_outdated_dag_access_control(value)
 
+    @property
+    def display_name(self) -> str | None:
+        return self._display_name
+
     @property
     def description(self) -> str | None:
         return self._description
@@ -2737,6 +2745,7 @@ def bulk_write_to_db(
             orm_dag.has_import_errors = False
             orm_dag.last_parsed_time = timezone.utcnow()
             orm_dag.default_view = dag.default_view
+            orm_dag.display_name = dag._display_name
             orm_dag.description = dag.description
             orm_dag.max_active_tasks = dag.max_active_tasks
             orm_dag.max_active_runs = dag.max_active_runs
@@ -3131,6 +3140,8 @@ class DagModel(Base):
     processor_subdir = Column(String(2000), nullable=True)
     # String representing the owners
     owners = Column(String(2000))
+    # Display name of the dag
+    display_name = Column(Text)
     # Description of the dag
     description = Column(Text)
     # Default view of the DAG inside the webserver

airflow/models/mappedoperator.py

@@ -253,6 +253,7 @@ class MappedOperator(AbstractOperator):
 
     # Needed for serialization.
     task_id: str
+    display_name: str
     params: ParamsDict | dict
     deps: frozenset[BaseTIDep]
     operator_extra_links: Collection[BaseOperatorLink]

airflow/models/taskinstance.py

@@ -398,6 +398,7 @@ class TaskInstance(Base, LoggingMixin):
     next_method = Column(String(1000))
     next_kwargs = Column(MutableDict.as_mutable(ExtendedJSON))
 
+    display_name = Column(Text)
     # If adding new fields here then remember to add them to
     # refresh_from_db() or they won't display in the UI correctly
 
@@ -527,6 +528,7 @@ def insert_mapping(run_id: str, task: Operator, map_index: int) -> dict[str, Any
             "executor_config": task.executor_config,
             "operator": task.task_type,
             "map_index": map_index,
+            "display_name": task.display_name,
         }
 
     @reconstructor
@@ -814,6 +816,7 @@ def refresh_from_db(self, session: Session = NEW_SESSION, lock_for_update: bool
             self.trigger_id = ti.trigger_id
             self.next_method = ti.next_method
             self.next_kwargs = ti.next_kwargs
+            self.display_name = ti.display_name
         else:
             self.state = None
 

airflow/serialization/schema.json

@@ -140,6 +140,7 @@
             ]
         },
          "orientation": { "type" : "string"},
+        "_display_name": { "type" : "string"},
         "_description": { "type" : "string"},
         "_concurrency": { "type" : "number"},
         "_max_active_tasks": { "type" : "number"},

airflow/serialization/serialized_objects.py

@@ -1137,6 +1137,7 @@ class SerializedDAG(DAG, BaseSerialization):
     def __get_constructor_defaults():
         param_to_attr = {
             "max_active_tasks": "_max_active_tasks",
+            "display_name": "_display_name",
             "description": "_description",
             "default_view": "_default_view",
             "access_control": "_access_control",

airflow/www/templates/airflow/dag.html

@@ -98,7 +98,7 @@
   {% if dag.parent_dag is defined and dag.parent_dag %}
     <a href="{{ url_for('Airflow.' + dag.default_view, dag_id=dag.parent_dag.dag_id, base_date=base_date_arg, execution_date=execution_date_arg) }}" title="Go to parent DAG">
       <span class="material-icons" aria-hidden="true">keyboard_arrow_up</span>
-      DAG: {{ dag.parent_dag.dag_id }}</a>
+      DAG: {{ dag.parent_dag.display_name }}</a>
   {% endif %}
 
   <div>
@@ -118,7 +118,7 @@ <h3 class="pull-left">
                  {{ " disabled" if not can_edit else "" }}>
           <span class="switch" aria-hidden="true"></span>
         </label>
-        <span class="text-muted">DAG:</span> {{ dag.dag_id }}
+        <span class="text-muted">DAG:</span> <span title="{{dag.dag_id}}">{{ dag.display_name }} </span>
         <small class="text-muted">{{ dag.description[0:150] + '…' if dag.description and dag.description|length > 150 else dag.description|default('', true) }}</small>
       {% endif %}
       {% if root %}

airflow/www/templates/airflow/dags.html

@@ -243,7 +243,7 @@ <h2>{{ page_title }}</h2>
             <td>
               <a href="{{ url_for('Airflow.'+ dag.get_default_view(), dag_id=dag.dag_id) }}"
                  title="{{ dag.description[0:80] + '…' if dag.description and dag.description|length > 80 else dag.description|default('', true) }}">
-                <strong>{{ dag.dag_id }}</strong>
+                <strong title="{{dag.dag_id}}">{{ dag.display_name }}</strong>
               </a>
               <div>
                 {% for tag in dag.tags | sort(attribute='name') %}

airflow/www/templates/airflow/task_instance.html

@@ -25,7 +25,7 @@
   <hr>
   <br>
   <h4>
-    <span class="text-muted">Task Instance:</span> <span>{{ task_id }}</span>
+    <span class="text-muted">Task Instance:</span> <span title="{{ task_id }}">{{display_name}}</span>
     <span class="text-muted">at</span> <time datetime="{{ execution_date }}">{{ execution_date }}</time>
     {% if map_index is defined and map_index >= 0 %}
       <span class="text-muted">Map Index:</span> <span>{{ map_index }}</span>

airflow/www/views.py

@@ -1713,6 +1713,7 @@ def include_task_attrs(attr_name):
             root=root,
             dag=dag,
             title=title,
+            display_name=task.display_name,
         )
 
     @expose("/xcom")