diff --git a/airflow/migrations/versions/0123_2_5_0_add_display_name_for_dag_and_task_.py b/airflow/migrations/versions/0123_2_5_0_add_display_name_for_dag_and_task_.py new file mode 100644 index 0000000000000..c701d233f544a --- /dev/null +++ b/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") diff --git a/airflow/models/baseoperator.py b/airflow/models/baseoperator.py index bb9b94b52f2aa..22a248db7f51a 100644 --- a/airflow/models/baseoperator.py +++ b/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() diff --git a/airflow/models/dag.py b/airflow/models/dag.py index c08a7f236f5d1..d196b7ab8419d 100644 --- a/airflow/models/dag.py +++ b/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 diff --git a/airflow/models/mappedoperator.py b/airflow/models/mappedoperator.py index 99e2b67f50018..2cdc1cf7fcec3 100644 --- a/airflow/models/mappedoperator.py +++ b/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] diff --git a/airflow/models/taskinstance.py b/airflow/models/taskinstance.py index 945e6b00a8450..e0b788188ac63 100644 --- a/airflow/models/taskinstance.py +++ b/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 diff --git a/airflow/serialization/schema.json b/airflow/serialization/schema.json index 19d6c4d5c9a4e..77086d086330f 100644 --- a/airflow/serialization/schema.json +++ b/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"}, diff --git a/airflow/serialization/serialized_objects.py b/airflow/serialization/serialized_objects.py index d6340c4b512a9..bc0bd45fe6bb4 100644 --- a/airflow/serialization/serialized_objects.py +++ b/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", diff --git a/airflow/www/templates/airflow/dag.html b/airflow/www/templates/airflow/dag.html index e6da97e68d5db..96617a31d8efb 100644 --- a/airflow/www/templates/airflow/dag.html +++ b/airflow/www/templates/airflow/dag.html @@ -98,7 +98,7 @@ {% if dag.parent_dag is defined and dag.parent_dag %} - DAG: {{ dag.parent_dag.dag_id }} + DAG: {{ dag.parent_dag.display_name }} {% endif %}