ashb commented on a change in pull request #22583:
URL: https://github.com/apache/airflow/pull/22583#discussion_r839508909
##########
File path: airflow/models/taskinstance.py
##########
@@ -2339,7 +2350,15 @@ def xcom_pull(
.order_by(None)
.order_by(XCom.map_index.asc())
)
- return (XCom.deserialize_value(r) for r in query)
+
+ def iter_xcom_values(query):
+ # The session passed to xcom_pull() may die before this is
+ # iterated through, so we need to bind to a new session.
+ with create_session() as session:
+ for r in query.with_session(session):
+ yield XCom.deserialize_value(r)
Review comment:
We shouldn't use `create_session` here as that has the side effect of
doing an expunge_all when this block exits.
We just want a `session = settings.Session()`
```suggestion
for r in query.with_session(settings.Session()):
yield XCom.deserialize_value(r)
```
--
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]
