msumit commented on code in PR #51076:
URL: https://github.com/apache/airflow/pull/51076#discussion_r2109669661
##########
airflow-core/src/airflow/dag_processing/manager.py:
##########
@@ -1141,14 +1142,42 @@ def process_parse_results(
stat.import_errors = 1
else:
# record DAGs and import errors to database
- update_dag_parsing_results_in_db(
- bundle_name=bundle_name,
- bundle_version=bundle_version,
- dags=parsing_result.serialized_dags,
- import_errors=parsing_result.import_errors or {},
- warnings=set(parsing_result.warnings or []),
- session=session,
- )
+ logging.info(f"Updating DAGs and import errors to database for bundle
{bundle_name} "
+ f"and version {bundle_version}")
+
+ # check a flag to see whether to call api or directly update db
+ if conf.getboolean("dag_processor", "use_api_for_updating_dags",
fallback=False):
+ server_url = conf.get("core", "execution_api_server_url")
+ if not server_url:
+ raise AirflowConfigException("execution_api_server_url is not
set")
+
+ logging.info("Attempting to make an API call to update DAGs using
the execution API server.")
+
+ @tenacity.retry(
+ stop=tenacity.stop_after_attempt(5),
+ wait=tenacity.wait_exponential(multiplier=1, min=4, max=15),
+ before_sleep=lambda retry_state: logging.info(
+ "Retrying update_dag_parsing_results_in_db. Attempt %d",
retry_state.attempt_number
+ ),
+ )
+ def _update_dags_via_api():
+ client = Client(base_url=server_url, token="")
+ client.post(
+
f"/dags/update_dags?bundle_name={bundle_name}&bundle_version={bundle_version}",
+ data=parsing_result.model_dump_json(),
+ headers={"Content-Type": "application/json"},
+ )
+ _update_dags_via_api()
Review Comment:
yeah may-be, but would've access to call these apis
--
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]
