pierrejeambrun commented on code in PR #51920:
URL: https://github.com/apache/airflow/pull/51920#discussion_r2166743303
##########
airflow-core/src/airflow/api_fastapi/core_api/routes/public/dag_run.py:
##########
@@ -432,6 +437,39 @@ def trigger_dag_run(
raise HTTPException(status.HTTP_400_BAD_REQUEST, str(e))
+async def _watch_dagrun(dag_id: str, run_id: str, interval: float) ->
AsyncGenerator[str, None]:
+ async with create_session_async() as session:
+ dag_run = await session.scalar(select(DagRun).filter_by(dag_id=dag_id,
run_id=run_id))
+ yield DAGRunWatchResult.model_validate(dag_run,
from_attributes=True).model_dump_json()
+ yield "\n"
+ while dag_run.state not in State.finished_dr_states:
+ await asyncio.sleep(interval)
+ async with create_session_async() as session:
+ dag_run = await
session.scalar(select(DagRun).filter_by(dag_id=dag_id, run_id=run_id))
+ yield DAGRunWatchResult.model_validate(dag_run,
from_attributes=True).model_dump_json()
+ yield "\n"
Review Comment:
This is duplicated from above, maybe should be extracted into a small fn.
Also we tend to not put services/utility code in the routes (endpoint) files
directly. Those can live in `services/public/dag_runs.py`
--
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: commits-unsubscr...@airflow.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
