wislertt opened a new issue, #73428:
URL: https://github.com/apache/airflow/issues/73428
## Apache Airflow version
3.3.2 (latest). I reproduced the bug with the DAG below on 3.3.1 and 3.3.2.
The same code is also present on `main`.
## What happened?
Running `airflow dags test` on a DAG that sets `deadline=DeadlineAlert(...)`
completes the run successfully (tasks succeed, DagRun is marked success), then
crashes with a traceback during the final `DagRun.update_state()` call:
```
Traceback (most recent call last):
...
File "airflow/cli/commands/dag_command.py", line 834, in dag_test
dr: DagRun = dag.test(...)
File "airflow/sdk/definitions/dag.py", line 1397, in test
schedulable_tis, _ = dr.update_state(session=session)
File "airflow/models/dagrun.py", line 1270, in update_state
DeadlineAlertModel.get_by_id(alert_id, session=session) for alert_id in
dag.deadline
File "airflow/models/deadline_alert.py", line 115, in get_by_id
result = session.scalar(select(cls).where(cls.id == deadline_alert_id))
...
File "sqlalchemy/sql/sqltypes.py", line 3738, in process
value = value.hex
AttributeError: 'dict' object has no attribute 'hex'
```
The exit code is non-zero, so the command reports failure for a run that
actually succeeded.
## Root cause
`DAG.test()` builds its "scheduler DAG" with an in-memory serialization
round trip:
```python
scheduler_dag =
DagSerialization.deserialize_dag(DagSerialization.serialize_dag(self))
```
(`airflow/sdk/definitions/dag.py`, `DAG.test()`)
Serialization encodes deadline alerts as plain dicts:
```python
serialized_dag["deadline"] = [encode_deadline_alert(d) for d in dag.deadline]
```
(`airflow/serialization/serialized_objects.py`, around line 1742)
but deserialization stores the field back without decoding it:
```python
dag.deadline = encoded_dag.get("deadline")
```
(`airflow/serialization/serialized_objects.py`, around line 1889)
So `scheduler_dag.deadline` is a `list` of encoded alert dicts.
When the run then succeeds, `DagRun.update_state()` prunes deadlines that
are no longer needed:
```python
deadline_alerts = [
DeadlineAlertModel.get_by_id(alert_id, session=session) for alert_id in
dag.deadline
]
```
(`airflow/models/dagrun.py`, around line 1270; same code around line 1371 on
`main`)
`get_by_id` expects `str | UUID`, but each element here is an encoded alert
dict, so SQLAlchemy's `Uuid` bind processor fails on `value.hex`.
The regular DB write path does not hit this because
`SerializedDagModel._generate_deadline_uuids`
(`airflow/models/serialized_dag.py`, around line 403) rewrites the serialized
`deadline` field from alert dicts into a list of `deadline_alert` UUID strings
before persisting. The in-memory round trip in `DAG.test()` bypasses that
conversion.
## How to reproduce
Save as `deadline_dagtest_repro.py` and run `airflow dags test
deadline_dagtest_repro`. Tested as-is on 3.3.1 and 3.3.2.
```python
from datetime import timedelta
from airflow.providers.standard.operators.empty import EmptyOperator
from airflow.sdk import DAG
from airflow.sdk.definitions.callback import SyncCallback
from airflow.sdk.definitions.deadline import DeadlineAlert, DeadlineReference
def _cb(**kwargs):
pass
with DAG(
dag_id="deadline_dagtest_repro",
schedule=None,
deadline=DeadlineAlert(
reference=DeadlineReference.DAGRUN_LOGICAL_DATE,
interval=timedelta(hours=2),
callback=SyncCallback(_cb),
),
):
EmptyOperator(task_id="task")
```
The task succeeds and the DagRun is marked success, then the traceback above
is raised and the CLI exits with a non-zero code.
## What you think should happen instead?
`airflow dags test` should complete cleanly (exit code 0) for DAGs with
deadlines. Some possible fixes:
1. Decode the deadline entries in `deserialize_dag` so `dag.deadline` holds
proper objects, matching how other serialized fields are decoded, or
2. Make the deadline pruning block in `DagRun.update_state()` tolerant of
both shapes (encoded dicts and UUID strings), or
3. Apply the same dict-to-UUID conversion that `write_dag` performs inside
the `DAG.test()` round trip (though no DB rows exist in the test path, so 1 or
2 is probably cleaner).
## Anything else?
- The scheduler is not affected, since it consumes the persisted serialized
DAG whose `deadline` field is already a list of UUID strings.
- Existing deadline PRs (#63701, #68195, #70148, #71628) all touch the DB
write path. None of them address the in-memory round trip in `DAG.test()`.
## Operating System
macOS (Darwin), Python 3.12 and 3.14. Not OS specific.
## Versions of Apache Airflow Providers
apache-airflow-providers-standard (installed from the constraints file of
each release).
## Deployment
Virtualenv install, local CLI testing. Not deployment specific.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
