Add logical_date to models using execution_date (#44283)

Airflow 3 renames all database columns from 'execution_date' to
'logical_date'. This backports the new name to 2.x so users can change
their code accessing those fields to use the new name before upgrading
to Airflow 3.

Author: uranusjr

airflow/models/log.py

@@ -26,6 +26,8 @@
 from airflow.utils.sqlalchemy import UtcDateTime
 
 if TYPE_CHECKING:
+    from datetime import datetime
+
     from airflow.models.taskinstance import TaskInstance
     from airflow.models.taskinstancekey import TaskInstanceKey
 
@@ -100,3 +102,7 @@ def __init__(
 
     def __str__(self) -> str:
         return f"Log({self.event}, {self.task_id}, {self.owner}, {self.owner_display_name}, {self.extra})"
+
+    @property
+    def logical_date(self) -> datetime:
+        return self.execution_date

airflow/models/renderedtifields.py

@@ -115,6 +115,7 @@ class RenderedTaskInstanceFields(TaskInstanceDependencies):
     )
 
     execution_date = association_proxy("dag_run", "execution_date")
+    logical_date = association_proxy("dag_run", "execution_date")
 
     def __init__(self, ti: TaskInstance, render_templates=True, rendered_fields=None):
         self.dag_id = ti.dag_id

airflow/models/slamiss.py

@@ -17,11 +17,16 @@
 # under the License.
 from __future__ import annotations
 
+from typing import TYPE_CHECKING
+
 from sqlalchemy import Boolean, Column, Index, String, Text
 
 from airflow.models.base import COLLATION_ARGS, ID_LEN, Base
 from airflow.utils.sqlalchemy import UtcDateTime
 
+if TYPE_CHECKING:
+    from datetime import datetime
+
 
 class SlaMiss(Base):
     """
@@ -44,3 +49,7 @@ class SlaMiss(Base):
 
     def __repr__(self):
         return str((self.dag_id, self.task_id, self.execution_date.isoformat()))
+
+    @property
+    def logical_date(self) -> datetime:
+        return self.execution_date

airflow/models/taskinstance.py

@@ -1902,6 +1902,7 @@ class TaskInstance(Base, LoggingMixin):
     dag_run = relationship("DagRun", back_populates="task_instances", lazy="joined", innerjoin=True)
     rendered_task_instance_fields = relationship("RenderedTaskInstanceFields", lazy="noload", uselist=False)
     execution_date = association_proxy("dag_run", "execution_date")
+    logical_date = association_proxy("dag_run", "execution_date")
     task_instance_note = relationship(
         "TaskInstanceNote",
         back_populates="task_instance",

airflow/models/taskreschedule.py

@@ -80,6 +80,7 @@ class TaskReschedule(TaskInstanceDependencies):
     )
     dag_run = relationship("DagRun")
     execution_date = association_proxy("dag_run", "execution_date")
+    logical_date = association_proxy("dag_run", "execution_date")
 
     def __init__(
         self,

airflow/models/xcom.py

@@ -119,6 +119,7 @@ class BaseXCom(TaskInstanceDependencies, LoggingMixin):
         passive_deletes="all",
     )
     execution_date = association_proxy("dag_run", "execution_date")
+    logical_date = association_proxy("dag_run", "execution_date")
 
     @reconstructor
     def init_on_load(self):

scripts/ci/pre_commit/check_ti_vs_tis_attributes.py

@@ -48,6 +48,7 @@ def compare_attributes(path1, path2):
         "task_instance_note",
         "dag_run",
         "trigger",
+        "execution_date",
         "logical_date",
         "triggerer_job",
         "note",