Repository: apache/airflow
File: airflow-core/src/airflow/callbacks/callback_requests.py (Python)

Code before:

class EmailRequest(BaseCallbackRequest):
    """Email notification request for task failures/retries."""

    ti: ti_datamodel.TaskInstance
    """Simplified Task Instance representation"""
    email_type: Literal["failure", "retry"] = "failure"
    """Whether this is for a failure or retry email"""
    context_from_server: ti_datamodel.TIRunContext
    """Task execution context from the Server"""
    type: Literal["EmailRequest"] = "EmailRequest"


class DagRunContext(BaseModel):
    """Class to pass context info from the server to build a Execution context object."""

    dag_run: ti_datamodel.DagRun | None = None
    last_ti: ti_datamodel.TaskInstance | None = None

    @model_validator(mode="before")
    @classmethod
    def _sanitize_consumed_asset_events(cls, values: Mapping[str, Any]) -> Mapping[str, Any]:
        dag_run = values.get("dag_run")

        if dag_run is None:
            return values

        # DagRunContext may receive non-ORM dag_run objects (e.g. datamodels).
        # Only apply this validator to ORM-mapped instances.
        try:
            _ = sa_inspect(dag_run)
        except NoInspectionAvailable:
            return values

        # Relationship access may raise DetachedInstanceError; fall back to empty list to
        # avoid crashing the scheduler.
        try:
            events = dag_run.consumed_asset_events
            set_committed_value(
                dag_run,
                "consumed_asset_events",
                list(events) if events is not None else [],
            )
        except DetachedInstanceError:
            log.warning(
                "DagRunContext encountered DetachedInstanceError while accessing "
                "consumed_asset_events; falling back to empty list"
            )
            set_committed_value(dag_run, "consumed_asset_events", [])

        return values

Review comment (suggestion):
```suggestion
        if (dag_run := values.get("dag_run")) is None:
            return values
```

Code after:

class EmailRequest(BaseCallbackRequest):
    """Email notification request for task failures/retries."""

    ti: ti_datamodel.TaskInstance
    """Simplified Task Instance representation"""
    email_type: Literal["failure", "retry"] = "failure"
    """Whether this is for a failure or retry email"""
    context_from_server: ti_datamodel.TIRunContext
    """Task execution context from the Server"""
    type: Literal["EmailRequest"] = "EmailRequest"


class DagRunContext(BaseModel):
    """Class to pass context info from the server to build a Execution context object."""

    dag_run: ti_datamodel.DagRun | None = None
    last_ti: ti_datamodel.TaskInstance | None = None

    @model_validator(mode="before")
    @classmethod
    def _sanitize_consumed_asset_events(cls, values: Mapping[str, Any]) -> Mapping[str, Any]:
        if (dag_run := values.get("dag_run")) is None:
            return values

        # DagRunContext may receive non-ORM dag_run objects (e.g. datamodels).
        # Only apply this validator to ORM-mapped instances.
        try:
            sa_inspect(dag_run)
        except NoInspectionAvailable:
            return values

        # Relationship access may raise DetachedInstanceError; on that path, reload DagRun
        # from the DB to avoid crashing the scheduler.
        try:
            events = dag_run.consumed_asset_events
            set_committed_value(
                dag_run,
                "consumed_asset_events",
                list(events) if events is not None else [],
            )
        except DetachedInstanceError:
            log.warning(
                "DagRunContext encountered DetachedInstanceError while accessing "
                "consumed_asset_events; reloading DagRun from DB."
            )
            from sqlalchemy import select
            from sqlalchemy.orm import selectinload

            from airflow.models.asset import AssetEvent
            from airflow.models.dagrun import DagRun