ephraimbuddy commented on a change in pull request #16182:
URL: https://github.com/apache/airflow/pull/16182#discussion_r657501138
##########
File path: airflow/models/dagbag.py
##########
@@ -166,6 +166,31 @@ def dag_ids(self) -> List[str]:
"""
return list(self.dags.keys())
+ @provide_session
+ def has_dag(self, dag_id: str, session: Session = None):
+ """
+ Checks the "local" cache, if it exists, and is not older than the
configured cache time, return True
+ else check in the DB if the dag exists, return True, False otherwise
+ """
+ from airflow.models.serialized_dag import SerializedDagModel
+
+ if dag_id in self.dags:
+ min_serialized_dag_fetch_secs =
timedelta(seconds=settings.MIN_SERIALIZED_DAG_FETCH_INTERVAL)
+ if (
+ dag_id in self.dags_last_fetched
+ and timezone.utcnow() > self.dags_last_fetched[dag_id] +
min_serialized_dag_fetch_secs
+ ):
+ sd_last_updated_datetime =
SerializedDagModel.get_last_updated_datetime(
+ dag_id=dag_id,
+ session=session,
+ )
+ if not sd_last_updated_datetime:
+ return False
+ return not sd_last_updated_datetime >
self.dags_last_fetched[dag_id]
+
+ return SerializedDagModel.has_dag(dag_id=dag_id, session=session)
Review comment:
It seems like this is needed for a case where the dag is in `self.dags`
but not in `self.dags_last_fetched`. Is it possible?
--
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.
For queries about this service, please contact Infrastructure at:
[email protected]
