MatrixManAtYrService opened a new issue, #26360:
URL: https://github.com/apache/airflow/issues/26360
### Apache Airflow version
2.4.0b1
### What happened
Here's a file which defines three dags. "source" uses `Operator.partial` to
reference either "sink". I'm not sure if it's supported to do so, but airlflow
should at least fail more gracefully than it does.
```python3
from datetime import datetime, timedelta
from time import sleep
from airflow import Dataset
from airflow.decorators import dag
from airflow.operators.dummy import DummyOperator
from airflow.operators.python import PythonOperator
ps = Dataset("partial_static")
p1 = Dataset("partial_dynamic_1")
p2 = Dataset("partial_dynamic_2")
p3 = Dataset("partial_dynamic_3")
def sleep_n(n):
sleep(n)
@dag(start_date=datetime(1970, 1, 1), schedule=timedelta(days=365 * 30))
def two_kinds_dynamic_source():
# dataset ref is not dynamic
PythonOperator.partial(
task_id="partial_static", python_callable=sleep_n, outlets=[ps]
).expand(op_args=[[1], [20], [40]])
# dataset ref is dynamic
PythonOperator.partial(
task_id="partial_dynamic", python_callable=sleep_n
).expand_kwargs(
[
{"op_args": [1], "outlets": [p1]},
{"op_args": [20], "outlets": [p2]},
{"op_args": [40], "outlets": [p3]},
]
)
two_kinds_dynamic_source()
@dag(schedule=[ps], start_date=datetime(1970, 1, 1))
def two_kinds_static_sink():
DummyOperator(task_id="dummy")
two_kinds_static_sink()
@dag(schedule=[p1, p2, p3], start_date=datetime(1970, 1, 1))
def two_kinds_dynamic_sink():
DummyOperator(task_id="dummy")
two_kinds_dynamic_sink()
```
Tried to view the "source" dag in the grid view, saw this traceback in
browser:
```
Python version: 3.9.13
Airflow version: 2.4.0.dev1640+astro.1
Node: airflow-webserver-6b969cbd87-4q5kh
-------------------------------------------------------------------------------
Traceback (most recent call last):
File "/usr/local/lib/python3.9/site-packages/flask/app.py", line 2525, in
wsgi_app
response = self.full_dispatch_request()
File "/usr/local/lib/python3.9/site-packages/flask/app.py", line 1822, in
full_dispatch_request
rv = self.handle_user_exception(e)
File "/usr/local/lib/python3.9/site-packages/flask/app.py", line 1820, in
full_dispatch_request
rv = self.dispatch_request()
File "/usr/local/lib/python3.9/site-packages/flask/app.py", line 1796, in
dispatch_request
return self.ensure_sync(self.view_functions[rule.endpoint])(**view_args)
File "/usr/local/lib/python3.9/site-packages/airflow/www/auth.py", line
46, in decorated
return func(*args, **kwargs)
File "/usr/local/lib/python3.9/site-packages/airflow/www/decorators.py",
line 117, in view_func
return f(*args, **kwargs)
File "/usr/local/lib/python3.9/site-packages/airflow/www/decorators.py",
line 80, in wrapper
return f(*args, **kwargs)
File "/usr/local/lib/python3.9/site-packages/airflow/utils/session.py",
line 73, in wrapper
return func(*args, session=session, **kwargs)
File "/usr/local/lib/python3.9/site-packages/airflow/www/views.py", line
2532, in grid
dag = get_airflow_app().dag_bag.get_dag(dag_id, session=session)
File "/usr/local/lib/python3.9/site-packages/airflow/utils/session.py",
line 70, in wrapper
return func(*args, **kwargs)
File "/usr/local/lib/python3.9/site-packages/airflow/models/dagbag.py",
line 176, in get_dag
self._add_dag_from_db(dag_id=dag_id, session=session)
File "/usr/local/lib/python3.9/site-packages/airflow/models/dagbag.py",
line 251, in _add_dag_from_db
dag = row.dag
File
"/usr/local/lib/python3.9/site-packages/airflow/models/serialized_dag.py", line
223, in dag
dag = SerializedDAG.from_dict(self.data) # type: Any
File
"/usr/local/lib/python3.9/site-packages/airflow/serialization/serialized_objects.py",
line 1220, in from_dict
return cls.deserialize_dag(serialized_obj['dag'])
File
"/usr/local/lib/python3.9/site-packages/airflow/serialization/serialized_objects.py",
line 1197, in deserialize_dag
setattr(task, k, kwargs_ref.deref(dag))
File
"/usr/local/lib/python3.9/site-packages/airflow/serialization/serialized_objects.py",
line 224, in deref
value = {k: v.deref(dag) if isinstance(v, _XComRef) else v for k, v in
self.value.items()}
AttributeError: 'list' object has no attribute 'items'
```
### What you think should happen instead
If there's something invalid about this dag, it should fail to parse--rather
than successfully parsing and then breaking the UI.
I'm a bit uncertain about what should happen in the dag dependency graph
when the source dag runs. The dynamic outlets are not known until runtime, so
it's reasonable that they don't show up in the graph. But what about after the
dag runs?
- do they still trigger the "sink" dag even though we didn't know about the
dependency up front?
- do we update the dependency graph now that we know about the dynamic
dependency?
Because of this error, we don't get far enough to find out.
### How to reproduce
Include the dag above, try to display it in the grid view.
### Operating System
kubernetes-in-docker on MacOS via helm
### Versions of Apache Airflow Providers
n/a
### Deployment
Other 3rd-party Helm chart
### Deployment details
Deployed using [the astronomer helm chart
](https://github.com/astronomer/airflow-chart)and these values:
```yaml
airflow:
airflowHome: /usr/local/airflow
airflowVersion: $VERSION
defaultAirflowRepository: img
defaultAirflowTag: $TAG
executor: KubernetesExecutor
gid: 50000
images:
airflow:
repository: img
logs:
persistence:
enabled: true
size: 2Gi
```
### Anything else
_No response_
### Are you willing to submit PR?
- [ ] Yes I am willing to submit a PR!
### Code of Conduct
- [X] I agree to follow this project's [Code of
Conduct](https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md)
--
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]
