You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "uranusjr (via GitHub)" <gi...@apache.org> on 2025/06/19 08:01:43 UTC
[PR] Add endpoint to watch dag run until finish [airflow]
uranusjr opened a new pull request, #51920:
URL: https://github.com/apache/airflow/pull/51920
Close #51711.
I initially wanted to just enhance the trigger endpoint to optionally stream until the run finishes, but it seems that FastAPI does not like this _optionally stream_ idea. You can do it of course, but would loose a lot of the auto annotation reflection feature. So I opted to have a separate streaming endpoint instead.
This endpoint repeatedly emits a JSON object at the specified interval, until the dag reaches a finished state.
--
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-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add endpoint to watch dag run until finish [airflow]
Posted by "pierrejeambrun (via GitHub)" <gi...@apache.org>.
pierrejeambrun commented on code in PR #51920:
URL: https://github.com/apache/airflow/pull/51920#discussion_r2182921964
##########
airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_dag_run.py:
##########
@@ -1606,3 +1606,49 @@ def test_should_respond_200_with_null_logical_date(self, test_client):
"conf": {},
"note": None,
}
+
+
+class TestWaitDagRun:
+ # The way we init async engine does not work well with FastAPI app init.
+ # Creating the engine implicitly creates an event loop, which Airflow does
+ # once for the entire process; creating the FastAPI app also does, but our
+ # test setup does it once for each test. I don't know how to properly fix
+ # this without rewriting how Airflow does db; re-configuring the db for each
+ # test at least makes the tests run correctly.
+ @pytest.fixture(autouse=True)
+ def reconfigure_async_db_engine(self):
+ from airflow.settings import _configure_async_session
+
+ _configure_async_session()
+
+ def test_should_respond_401(self, unauthenticated_test_client):
+ response = unauthenticated_test_client.get(f"/dags/{DAG1_ID}/dagRuns/{DAG1_RUN1_ID}/wait?interval=1")
+ assert response.status_code == 401
+
+ def test_should_respond_403(self, unauthorized_test_client):
+ response = unauthorized_test_client.get(f"/dags/{DAG1_ID}/dagRuns/{DAG1_RUN1_ID}/wait?interval=1")
+ assert response.status_code == 403
+
+ def test_should_respond_404(self, test_client):
+ response = test_client.get(f"/dags/{DAG1_ID}/dagRuns/does-not-exist/wait?interval=1")
+ assert response.status_code == 404
+
+ def test_should_respond_422_without_interval_param(self, test_client):
+ response = test_client.get(f"/dags/{DAG1_ID}/dagRuns/{DAG1_RUN1_ID}/wait")
+ assert response.status_code == 422
+
+ @pytest.mark.parametrize(
+ "run_id, state",
+ [(DAG1_RUN1_ID, DAG1_RUN1_STATE), (DAG1_RUN2_ID, DAG1_RUN2_STATE)],
+ )
+ def test_should_respond_200_immediately_for_finished_run(self, test_client, run_id, state):
+ response = test_client.get(f"/dags/{DAG1_ID}/dagRuns/{run_id}/wait?interval=100")
+ assert response.status_code == 200
+ data = response.json()
+ assert data == {"state": state}
+
+ def test_collect_task(self, test_client):
+ response = test_client.get(f"/dags/{DAG1_ID}/dagRuns/{DAG1_RUN1_ID}/wait?interval=100&collect=task_1")
Review Comment:
same here
##########
airflow-core/src/airflow/api_fastapi/core_api/openapi/v2-rest-api-generated.yaml:
##########
@@ -2135,6 +2135,87 @@ paths:
application/json:
schema:
$ref: '#/components/schemas/HTTPValidationError'
+ /api/v2/dags/{dag_id}/dagRuns/{dag_run_id}/wait:
Review Comment:
```suggestion
/api/v2/dags/{dag_id}/dagRuns/{dag_run_id}/watch:
```
Nit: To me it looks like we're watching the dag run state, and stop watching when it's settled. But I'm good with wait too.
##########
airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_dag_run.py:
##########
@@ -1606,3 +1606,49 @@ def test_should_respond_200_with_null_logical_date(self, test_client):
"conf": {},
"note": None,
}
+
+
+class TestWaitDagRun:
+ # The way we init async engine does not work well with FastAPI app init.
+ # Creating the engine implicitly creates an event loop, which Airflow does
+ # once for the entire process; creating the FastAPI app also does, but our
+ # test setup does it once for each test. I don't know how to properly fix
+ # this without rewriting how Airflow does db; re-configuring the db for each
+ # test at least makes the tests run correctly.
+ @pytest.fixture(autouse=True)
+ def reconfigure_async_db_engine(self):
+ from airflow.settings import _configure_async_session
+
+ _configure_async_session()
+
+ def test_should_respond_401(self, unauthenticated_test_client):
+ response = unauthenticated_test_client.get(f"/dags/{DAG1_ID}/dagRuns/{DAG1_RUN1_ID}/wait?interval=1")
+ assert response.status_code == 401
+
+ def test_should_respond_403(self, unauthorized_test_client):
+ response = unauthorized_test_client.get(f"/dags/{DAG1_ID}/dagRuns/{DAG1_RUN1_ID}/wait?interval=1")
+ assert response.status_code == 403
+
+ def test_should_respond_404(self, test_client):
+ response = test_client.get(f"/dags/{DAG1_ID}/dagRuns/does-not-exist/wait?interval=1")
+ assert response.status_code == 404
+
+ def test_should_respond_422_without_interval_param(self, test_client):
+ response = test_client.get(f"/dags/{DAG1_ID}/dagRuns/{DAG1_RUN1_ID}/wait")
+ assert response.status_code == 422
+
+ @pytest.mark.parametrize(
+ "run_id, state",
+ [(DAG1_RUN1_ID, DAG1_RUN1_STATE), (DAG1_RUN2_ID, DAG1_RUN2_STATE)],
+ )
+ def test_should_respond_200_immediately_for_finished_run(self, test_client, run_id, state):
+ response = test_client.get(f"/dags/{DAG1_ID}/dagRuns/{run_id}/wait?interval=100")
Review Comment:
Nit: Here and bellow, can you use `params=....` argument and let the test_client do the url encoding instead of doing this manually please.
It's easier to read especially when there are multiple query params (so we tend to do that everywhere), and manipulating a dict is more convenient that a string. (looking up variables, etc.)
--
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-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add endpoint to watch dag run until finish [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on PR #51920:
URL: https://github.com/apache/airflow/pull/51920#issuecomment-3051321348
Tracking doc addition
--
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-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add endpoint to watch dag run until finish [airflow]
Posted by "kaxil (via GitHub)" <gi...@apache.org>.
kaxil commented on code in PR #51920:
URL: https://github.com/apache/airflow/pull/51920#discussion_r2193154704
##########
airflow-core/src/airflow/api_fastapi/core_api/routes/public/dag_run.py:
##########
@@ -433,6 +437,57 @@ def trigger_dag_run(
raise HTTPException(status.HTTP_400_BAD_REQUEST, str(e))
+@dag_run_router.get(
+ "/{dag_run_id}/wait",
+ tags=["experimental"],
+ summary="Experimental: Wait for a dag run to complete, and return task results if requested.",
+ description="🚧 This is an experimental endpoint and may change or be removed without notice.",
+ responses={
+ **create_openapi_http_exception_doc([status.HTTP_404_NOT_FOUND]),
+ status.HTTP_200_OK: {
+ "description": "Successful Response",
+ "content": {
+ Mimetype.NDJSON: {
+ "schema": {
+ "type": "string",
+ "example": textwrap.dedent(
+ """\
+ {"state": "running"}
+ {"state": "success", "returns": {"op": 42}}
+ """
+ ),
+ }
+ }
+ },
+ },
+ },
+ dependencies=[Depends(requires_access_dag(method="GET", access_entity=DagAccessEntity.RUN))],
+)
+def wait_dag_run_until_finished(
+ dag_id: str,
+ dag_run_id: str,
+ session: SessionDep,
+ interval: Annotated[float, Query(gt=0.0, description="Seconds to wait between dag run state checks")],
+ collect_task_ids: Annotated[
+ list[str] | None,
+ Query(alias="collect", description="Collect return value XCom from task. Can be set multiple times."),
Review Comment:
`result` works as well. `result_task_id` works too. Your call @uranusjr
--
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-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add endpoint to watch dag run until finish [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #51920:
URL: https://github.com/apache/airflow/pull/51920#discussion_r2168148970
##########
airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_dag_run.py:
##########
@@ -1602,3 +1602,45 @@ def test_should_respond_200_with_null_logical_date(self, test_client):
"conf": {},
"note": None,
}
+
+
+class TestWatchDagRun:
+ # The way we init async engine does not work well with FastAPI app init.
+ # Creating the engine implicitly creates an event loop, which Airflow does
+ # once for the entire process; creating the FastAPI app also does, but our
+ # test setup does it once for each test. I don't know how to properly fix
+ # this without rewriting how Airflow does db; re-configuring the db for each
+ # test at least makes the tests run correctly.
+ @pytest.fixture(autouse=True)
+ def reconfigure_async_db_engine(self):
+ from airflow.settings import _configure_async_session
+
+ _configure_async_session()
Review Comment:
The `test_should_respond_200_immediately_for_finished_run` tests fail for me without this (there are two fixtures; the first would pass but the second would fail)
--
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-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add endpoint to watch dag run until finish [airflow]
Posted by "pierrejeambrun (via GitHub)" <gi...@apache.org>.
pierrejeambrun commented on code in PR #51920:
URL: https://github.com/apache/airflow/pull/51920#discussion_r2166723553
##########
airflow-core/src/airflow/api_fastapi/core_api/datamodels/dag_run.py:
##########
@@ -163,3 +163,10 @@ class DAGRunsBatchBody(StrictBaseModel):
start_date_lte: AwareDatetime | None = None
end_date_gte: AwareDatetime | None = None
end_date_lte: AwareDatetime | None = None
+
+
+class DAGRunWatchResult(StrictBaseModel):
Review Comment:
Do we really need a new Serializer? I'm not fan of having each endpoint returning custom schema, unless that would really impact / slow down performances. Here can't we just use the default `DAGRunResponse`, which holds the state and duration. This will be much easier for client to understand the API if we avoid multiplication of partial schemas.
##########
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/dag_run.py`
##########
airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_dag_run.py:
##########
@@ -1602,3 +1602,45 @@ def test_should_respond_200_with_null_logical_date(self, test_client):
"conf": {},
"note": None,
}
+
+
+class TestWatchDagRun:
+ # The way we init async engine does not work well with FastAPI app init.
+ # Creating the engine implicitly creates an event loop, which Airflow does
+ # once for the entire process; creating the FastAPI app also does, but our
+ # test setup does it once for each test. I don't know how to properly fix
+ # this without rewriting how Airflow does db; re-configuring the db for each
+ # test at least makes the tests run correctly.
+ @pytest.fixture(autouse=True)
+ def reconfigure_async_db_engine(self):
+ from airflow.settings import _configure_async_session
+
+ _configure_async_session()
Review Comment:
Is that still necessary? Tests are green for me even when removing this fixture.
--
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-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add endpoint to watch dag run until finish [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on PR #51920:
URL: https://github.com/apache/airflow/pull/51920#issuecomment-2987587459
FYI. Not sure if it helps but OpenAPI 3.2.0 will have support for `application/json-seq` https://github.com/OAI/OpenAPI-Specification/issues/3730 (planned for August)
--
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-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add endpoint to watch dag run until finish [airflow]
Posted by "kaxil (via GitHub)" <gi...@apache.org>.
kaxil commented on code in PR #51920:
URL: https://github.com/apache/airflow/pull/51920#discussion_r2190941717
##########
airflow-core/src/airflow/api_fastapi/core_api/routes/public/dag_run.py:
##########
@@ -433,6 +437,57 @@ def trigger_dag_run(
raise HTTPException(status.HTTP_400_BAD_REQUEST, str(e))
+@dag_run_router.get(
+ "/{dag_run_id}/wait",
+ tags=["experimental"],
+ summary="Experimental: Wait for a dag run to complete, and return task results if requested.",
+ description="🚧 This is an experimental endpoint and may change or be removed without notice.",
+ responses={
+ **create_openapi_http_exception_doc([status.HTTP_404_NOT_FOUND]),
+ status.HTTP_200_OK: {
+ "description": "Successful Response",
+ "content": {
+ Mimetype.NDJSON: {
+ "schema": {
+ "type": "string",
+ "example": textwrap.dedent(
+ """\
+ {"state": "running"}
+ {"state": "success", "returns": {"op": 42}}
+ """
+ ),
+ }
+ }
+ },
+ },
+ },
+ dependencies=[Depends(requires_access_dag(method="GET", access_entity=DagAccessEntity.RUN))],
+)
+def wait_dag_run_until_finished(
+ dag_id: str,
+ dag_run_id: str,
+ session: SessionDep,
+ interval: Annotated[float, Query(gt=0.0, description="Seconds to wait between dag run state checks")],
+ collect_task_ids: Annotated[
+ list[str] | None,
+ Query(alias="collect", description="Collect return value XCom from task. Can be set multiple times."),
Review Comment:
naming is hard: `collect` doesn't feel great. `result_task_ids`?
--
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-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add endpoint to watch dag run until finish [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #51920:
URL: https://github.com/apache/airflow/pull/51920#discussion_r2191495048
##########
airflow-core/src/airflow/api_fastapi/core_api/routes/public/dag_run.py:
##########
@@ -433,6 +437,57 @@ def trigger_dag_run(
raise HTTPException(status.HTTP_400_BAD_REQUEST, str(e))
+@dag_run_router.get(
+ "/{dag_run_id}/wait",
+ tags=["experimental"],
+ summary="Experimental: Wait for a dag run to complete, and return task results if requested.",
+ description="🚧 This is an experimental endpoint and may change or be removed without notice.",
+ responses={
+ **create_openapi_http_exception_doc([status.HTTP_404_NOT_FOUND]),
+ status.HTTP_200_OK: {
+ "description": "Successful Response",
+ "content": {
+ Mimetype.NDJSON: {
+ "schema": {
+ "type": "string",
+ "example": textwrap.dedent(
+ """\
+ {"state": "running"}
+ {"state": "success", "returns": {"op": 42}}
+ """
+ ),
+ }
+ }
+ },
+ },
+ },
+ dependencies=[Depends(requires_access_dag(method="GET", access_entity=DagAccessEntity.RUN))],
+)
+def wait_dag_run_until_finished(
+ dag_id: str,
+ dag_run_id: str,
+ session: SessionDep,
+ interval: Annotated[float, Query(gt=0.0, description="Seconds to wait between dag run state checks")],
+ collect_task_ids: Annotated[
+ list[str] | None,
+ Query(alias="collect", description="Collect return value XCom from task. Can be set multiple times."),
Review Comment:
How about the query key in the URL? We should use singular there since it’ll be used like this
```
https://.../wait?interval=100&result_task_id=op1&result_task_id=op2
```
This is a bit long to me, but not unacceptable. Would `result` work?
--
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-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add endpoint to watch dag run until finish [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on PR #51920:
URL: https://github.com/apache/airflow/pull/51920#issuecomment-3003369257
I added some tests for the endpoint, but couldn’t figure out how to test the looping part. Hopefully this is good enough…
--
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-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add endpoint to watch dag run until finish [airflow]
Posted by "pierrejeambrun (via GitHub)" <gi...@apache.org>.
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-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add endpoint to watch dag run until finish [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr merged PR #51920:
URL: https://github.com/apache/airflow/pull/51920
--
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-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add endpoint to watch dag run until finish [airflow]
Posted by "pierrejeambrun (via GitHub)" <gi...@apache.org>.
pierrejeambrun commented on code in PR #51920:
URL: https://github.com/apache/airflow/pull/51920#discussion_r2192480400
##########
airflow-core/src/airflow/api_fastapi/core_api/routes/public/dag_run.py:
##########
@@ -433,6 +437,57 @@ def trigger_dag_run(
raise HTTPException(status.HTTP_400_BAD_REQUEST, str(e))
+@dag_run_router.get(
+ "/{dag_run_id}/wait",
+ tags=["experimental"],
+ summary="Experimental: Wait for a dag run to complete, and return task results if requested.",
+ description="🚧 This is an experimental endpoint and may change or be removed without notice.",
+ responses={
+ **create_openapi_http_exception_doc([status.HTTP_404_NOT_FOUND]),
+ status.HTTP_200_OK: {
+ "description": "Successful Response",
+ "content": {
+ Mimetype.NDJSON: {
+ "schema": {
+ "type": "string",
+ "example": textwrap.dedent(
+ """\
+ {"state": "running"}
+ {"state": "success", "returns": {"op": 42}}
+ """
+ ),
+ }
+ }
+ },
+ },
+ },
+ dependencies=[Depends(requires_access_dag(method="GET", access_entity=DagAccessEntity.RUN))],
+)
+def wait_dag_run_until_finished(
+ dag_id: str,
+ dag_run_id: str,
+ session: SessionDep,
+ interval: Annotated[float, Query(gt=0.0, description="Seconds to wait between dag run state checks")],
+ collect_task_ids: Annotated[
+ list[str] | None,
+ Query(alias="collect", description="Collect return value XCom from task. Can be set multiple times."),
Review Comment:
+1 for singular, this is what we do in other endpoints.
No specific opinion on the naming though.
--
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-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add endpoint to watch dag run until finish [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #51920:
URL: https://github.com/apache/airflow/pull/51920#discussion_r2194068145
##########
airflow-core/src/airflow/api_fastapi/core_api/routes/public/dag_run.py:
##########
@@ -433,6 +437,57 @@ def trigger_dag_run(
raise HTTPException(status.HTTP_400_BAD_REQUEST, str(e))
+@dag_run_router.get(
+ "/{dag_run_id}/wait",
+ tags=["experimental"],
+ summary="Experimental: Wait for a dag run to complete, and return task results if requested.",
+ description="🚧 This is an experimental endpoint and may change or be removed without notice.",
+ responses={
+ **create_openapi_http_exception_doc([status.HTTP_404_NOT_FOUND]),
+ status.HTTP_200_OK: {
+ "description": "Successful Response",
+ "content": {
+ Mimetype.NDJSON: {
+ "schema": {
+ "type": "string",
+ "example": textwrap.dedent(
+ """\
+ {"state": "running"}
+ {"state": "success", "returns": {"op": 42}}
+ """
+ ),
+ }
+ }
+ },
+ },
+ },
+ dependencies=[Depends(requires_access_dag(method="GET", access_entity=DagAccessEntity.RUN))],
+)
+def wait_dag_run_until_finished(
+ dag_id: str,
+ dag_run_id: str,
+ session: SessionDep,
+ interval: Annotated[float, Query(gt=0.0, description="Seconds to wait between dag run state checks")],
+ collect_task_ids: Annotated[
+ list[str] | None,
+ Query(alias="collect", description="Collect return value XCom from task. Can be set multiple times."),
Review Comment:
Changed to `result`. Also changed the key used in the response to `results` to match.
--
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-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add endpoint to watch dag run until finish [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #51920:
URL: https://github.com/apache/airflow/pull/51920#discussion_r2185475808
##########
airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_dag_run.py:
##########
@@ -1606,3 +1606,49 @@ def test_should_respond_200_with_null_logical_date(self, test_client):
"conf": {},
"note": None,
}
+
+
+class TestWaitDagRun:
+ # The way we init async engine does not work well with FastAPI app init.
+ # Creating the engine implicitly creates an event loop, which Airflow does
+ # once for the entire process; creating the FastAPI app also does, but our
+ # test setup does it once for each test. I don't know how to properly fix
+ # this without rewriting how Airflow does db; re-configuring the db for each
+ # test at least makes the tests run correctly.
+ @pytest.fixture(autouse=True)
+ def reconfigure_async_db_engine(self):
+ from airflow.settings import _configure_async_session
+
+ _configure_async_session()
+
+ def test_should_respond_401(self, unauthenticated_test_client):
+ response = unauthenticated_test_client.get(f"/dags/{DAG1_ID}/dagRuns/{DAG1_RUN1_ID}/wait?interval=1")
+ assert response.status_code == 401
+
+ def test_should_respond_403(self, unauthorized_test_client):
+ response = unauthorized_test_client.get(f"/dags/{DAG1_ID}/dagRuns/{DAG1_RUN1_ID}/wait?interval=1")
+ assert response.status_code == 403
+
+ def test_should_respond_404(self, test_client):
+ response = test_client.get(f"/dags/{DAG1_ID}/dagRuns/does-not-exist/wait?interval=1")
+ assert response.status_code == 404
+
+ def test_should_respond_422_without_interval_param(self, test_client):
+ response = test_client.get(f"/dags/{DAG1_ID}/dagRuns/{DAG1_RUN1_ID}/wait")
+ assert response.status_code == 422
+
+ @pytest.mark.parametrize(
+ "run_id, state",
+ [(DAG1_RUN1_ID, DAG1_RUN1_STATE), (DAG1_RUN2_ID, DAG1_RUN2_STATE)],
+ )
+ def test_should_respond_200_immediately_for_finished_run(self, test_client, run_id, state):
+ response = test_client.get(f"/dags/{DAG1_ID}/dagRuns/{run_id}/wait?interval=100")
Review Comment:
I didn’t know you can do that! Definitely will be easier to read.
--
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-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org