pierrejeambrun commented on code in PR #43875:
URL: https://github.com/apache/airflow/pull/43875#discussion_r1858207506
##########
airflow/api_fastapi/core_api/routes/public/dag_run.py:
##########
@@ -296,3 +302,71 @@ def get_dag_runs(
dag_runs=dag_runs,
total_entries=total_entries,
)
+
+
+@dag_run_router.post(
+ "",
+ responses=create_openapi_http_exception_doc(
+ [
+ status.HTTP_400_BAD_REQUEST,
+ status.HTTP_404_NOT_FOUND,
+ status.HTTP_409_CONFLICT,
+ ]
+ ),
+)
+def trigger_dag_run(
+ dag_id, body: TriggerDAGRunPostBody, request: Request, session:
Annotated[Session, Depends(get_session)]
+) -> DAGRunResponse:
+ """Trigger a DAG."""
+ dm = session.scalar(select(DagModel).where(DagModel.is_active,
DagModel.dag_id == dag_id).limit(1))
+ if not dm:
+ raise HTTPException(status.HTTP_404_NOT_FOUND, f"DAG with dag_id:
'{dag_id}' not found")
+
+ if dm.has_import_errors:
+ raise HTTPException(
+ status.HTTP_400_BAD_REQUEST,
+ f"DAG with dag_id: '{dag_id}' has import errors and cannot be
triggered",
+ )
+
+ run_id = body.dag_run_id
+ logical_date = pendulum.instance(body.logical_date)
+ dagrun_instance = session.scalar(
+ select(DagRun).where(DagRun.dag_id == dag_id, DagRun.run_id ==
run_id).limit(1)
+ )
+
+ if not dagrun_instance:
+ try:
+ dag: DAG = request.app.state.dag_bag.get_dag(dag_id)
+
+ if body.data_interval_start and body.data_interval_end:
+ data_interval = DataInterval(
+ start=pendulum.instance(body.data_interval_start),
+ end=pendulum.instance(body.data_interval_end),
+ )
+ else:
+ data_interval =
dag.timetable.infer_manual_data_interval(run_after=logical_date)
+ dag_version = DagVersion.get_latest_version(dag.dag_id)
+ dag_run = dag.create_dagrun(
+ run_type=DagRunType.MANUAL,
+ run_id=run_id,
+ logical_date=logical_date,
+ data_interval=data_interval,
+ state=DagRunState.QUEUED,
+ conf=body.conf,
+ external_trigger=True,
+ dag_version=dag_version,
+ session=session,
+ triggered_by=DagRunTriggeredByType.REST_API,
+ )
+ dag_run_note = body.note
+ if dag_run_note:
+ current_user_id = None # refer to
https://github.com/apache/airflow/issues/43534
+ dag_run.note = (dag_run_note, current_user_id)
+ return dag_run
+ except ValueError as e:
+ raise HTTPException(status.HTTP_400_BAD_REQUEST, str(e))
+
+ raise HTTPException(
+ status.HTTP_409_CONFLICT,
+ f"DAGRun with DAG ID: '{dag_id}' and DAGRun ID: '{body.dag_run_id}'
already exists",
+ )
Review Comment:
There is currently a discussion on that here:
https://github.com/apache/airflow/pull/44322
Ephraim also find the message to broad and we would like to make it more
descriptive. In the meantime I think we should be consistent. We cannot have
one endpoint relying on the general exception handler for Integrity error
(returning one kind of message), and then have another endpoint doing its own
error handling throwing another kind of message.
I think the direction is most likely to improve the exception handler so
every endpoints benefit from it without having to add extra code to every
endpoint.
I would rely on this in this PR too, and let the other PR take care of
improving the message. (Or a successive PR).
--
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]
