ashb commented on a change in pull request #4328: AIRFLOW-3530 - No re-parsing
of dag files
URL: https://github.com/apache/incubator-airflow/pull/4328#discussion_r242232376
##########
File path: airflow/utils/dag_processing.py
##########
@@ -563,29 +442,31 @@ def helper():
p.start()
return p
- def harvest_simple_dags(self):
+ def harvest_dags(self):
"""
Harvest DAG parsing results from result queue and sync metadata from
stat queue.
- :return: List of parsing result in SimpleDag format.
+ :return: DagBag
"""
# Metadata and results to be harvested can be inconsistent,
# but it should not be a big problem.
self._sync_metadata()
# Heartbeating after syncing metadata so we do not restart manager
# if it processed all files for max_run times and exit normally.
self._heartbeat_manager()
- simple_dags = []
+ dags = []
# multiprocessing.Queue().qsize will not work on MacOS.
if sys.platform == "darwin":
qsize = self._result_count
else:
qsize = self._result_queue.qsize()
for _ in range(qsize):
- simple_dags.append(self._result_queue.get())
+ dags.append(self._result_queue.get())
self._result_count = 0
- return simple_dags
+ for dag in dags:
+ self.dag_bag.dags[dag.dag_id] = dag
Review comment:
I think there's a `dag_bag.bag_dag()` method that we should use instead?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
With regards,
Apache Git Services
