Bad request when triggering dag run with `note` in payload

[ttben]

Apache Airflow version

2.5.0

What happened

Specifying a note in the payload (as mentioned in the doc) when triggering a new dag run yield a 400 bad request

(Git Version: .release:2.5.0+fa2bec042995004f45b914dd1d66b466ccced410)

What you think should happen instead

As far as I understand the documentation, I should be able to set a note for this dag run, and it is not the case.

How to reproduce

This is a local airflow, using default credentials and default setup when following this guide

DAG:

Details

import airflow
from airflow import DAG

import logging

from airflow.operators.python import PythonOperator
from airflow.operators.dummy import DummyOperator

from datetime import timedelta

logger = logging.getLogger("airflow.task")

default_args = {
    "owner": "airflow",
    "depends_on_past": False,
    "retries": 0,
    "retry_delay": timedelta(minutes=5),
}

def log_body(**context):
    logger.info(f"Body: {context['dag_run'].conf}")


with DAG(
        "my-validator",
        default_args=default_args,
        schedule_interval=None,
        start_date=airflow.utils.dates.days_ago(0),
        catchup=False
) as dag:
    (
        PythonOperator(
            task_id="abcde",
            python_callable=log_body,
            provide_context=True
        )
        >> DummyOperator(
            task_id="todo"
        )
    )

Request:

Details

curl --location --request POST '0.0.0.0:8080/api/v1/dags/my-validator/dagRuns' \
--header 'Authorization: Basic YWlyZmxvdzphaXJmbG93' \
--header 'Content-Type: application/json' \
--data-raw '{
  "conf": {
      "key":"value"
  },
  "note": "test"
  }' 

Response:

Details

{
    "detail": "{'note': ['Unknown field.']}",
    "status": 400,
    "title": "Bad Request",
    "type": "https://airflow.apache.org/docs/apache-airflow/2.5.0/stable-rest-api-ref.html#section/Errors/BadRequest"
}

Removing the note key, returns 200... with a null note!

Details

{
    "conf": {
        "key": "value"
    },
    "dag_id": "my-validator",
    "dag_run_id": "manual__2023-01-10T10:45:26.102802+00:00",
    "data_interval_end": "2023-01-10T10:45:26.102802+00:00",
    "data_interval_start": "2023-01-10T10:45:26.102802+00:00",
    "end_date": null,
    "execution_date": "2023-01-10T10:45:26.102802+00:00",
    "external_trigger": true,
    "last_scheduling_decision": null,
    "logical_date": "2023-01-10T10:45:26.102802+00:00",
    "note": null,
    "run_type": "manual",
    "start_date": null,
    "state": "queued"
}

Operating System

Ubuntu 20.04.5 LTS

Versions of Apache Airflow Providers

No response

Deployment

Docker-Compose

Deployment details

No response

Anything else

Everytime.

Are you willing to submit PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct

[boring-cyborg]

Thanks for opening your first issue here! Be sure to follow the issue template!

[uranusjr]

The note field is incorrectly marked as dump_only=True. Feel free to contribute.

airflow/airflow/api_connexion/schemas/dag_run_schema.py

Line 76 in d5ac1b0

note = auto_field(dump_only=True)

[obilalk]

In v1.yaml you can see setted get-post request for airflow /dags/{dag_id}/dagRuns:

airflow/airflow/api_connexion/openapi/v1.yaml

Lines 700 to 759 in d5ac1b0

	/dags/{dag_id}/dagRuns:
	parameters:
	- $ref: '#/components/parameters/DAGID'
	get:
	summary: List DAG runs
	description: >
	This endpoint allows specifying `~` as the dag_id to retrieve DAG runs for all DAGs.
	x-openapi-router-controller: airflow.api_connexion.endpoints.dag_run_endpoint
	operationId: get_dag_runs
	tags: [DAGRun]
	parameters:
	- $ref: '#/components/parameters/PageLimit'
	- $ref: '#/components/parameters/PageOffset'
	- $ref: '#/components/parameters/FilterExecutionDateGTE'
	- $ref: '#/components/parameters/FilterExecutionDateLTE'
	- $ref: '#/components/parameters/FilterStartDateGTE'
	- $ref: '#/components/parameters/FilterStartDateLTE'
	- $ref: '#/components/parameters/FilterEndDateGTE'
	- $ref: '#/components/parameters/FilterEndDateLTE'
	- $ref: '#/components/parameters/FilterState'
	- $ref: '#/components/parameters/OrderBy'
	responses:
	'200':
	description: List of DAG runs.
	content:
	application/json:
	schema:
	$ref: '#/components/schemas/DAGRunCollection'
	'401':
	$ref: '#/components/responses/Unauthenticated'
	post:
	summary: Trigger a new DAG run
	x-openapi-router-controller: airflow.api_connexion.endpoints.dag_run_endpoint
	operationId: post_dag_run
	tags: [DAGRun]
	requestBody:
	required: true
	content:
	application/json:
	schema:
	$ref: '#/components/schemas/DAGRun'
	responses:
	'200':
	description: Success.
	content:
	application/json:
	schema:
	$ref: '#/components/schemas/DAGRun'
	'400':
	$ref: '#/components/responses/BadRequest'
	'401':
	$ref: '#/components/responses/Unauthenticated'
	'409':
	$ref: '#/components/responses/AlreadyExists'
	'403':
	$ref: '#/components/responses/PermissionDenied'
	'404':
	$ref: '#/components/responses/NotFound'

post_dag_run function is the main function to external trigger for Dag, in this case if dagrun_instance return False start to send payload to trigger airflow dag but there are no exist note section.

airflow/airflow/api_connexion/endpoints/dag_run_endpoint.py

Lines 308 to 325 in d5ac1b0

	if not dagrun_instance:
	try:
	dag = get_airflow_app().dag_bag.get_dag(dag_id)
	dag_run = dag.create_dagrun(
	run_type=DagRunType.MANUAL,
	run_id=run_id,
	execution_date=logical_date,
	data_interval=dag.timetable.infer_manual_data_interval(run_after=logical_date),
	state=DagRunState.QUEUED,
	conf=post_body.get("conf"),
	external_trigger=True,
	dag_hash=get_airflow_app().dag_bag.dags_hash.get(dag_id),
	session=session,
	)
	return dagrun_schema.dump(dag_run)
	except ValueError as ve:
	raise BadRequest(detail=str(ve))

On the other hand to set a note for DagRun with api call we need dagrun_instance True but it will conflict in post_dag_run function.

airflow/airflow/api_connexion/endpoints/dag_run_endpoint.py

Lines 426 to 450 in d5ac1b0

	@provide_session
	def set_dag_run_note(*, dag_id: str, dag_run_id: str, session: Session = NEW_SESSION) -> APIResponse:
	"""Set the note for a dag run."""
	dag_run: DagRun | None = (
	session.query(DagRun).filter(DagRun.dag_id == dag_id, DagRun.run_id == dag_run_id).one_or_none()
	)
	if dag_run is None:
	error_message = f"Dag Run id {dag_run_id} not found in dag {dag_id}"
	raise NotFound(error_message)
	try:
	post_body = set_dagrun_note_form_schema.load(get_json_request_dict())
	new_note = post_body["note"]
	except ValidationError as err:
	raise BadRequest(detail=str(err))
	from flask_login import current_user
	current_user_id = getattr(current_user, "id", None)
	if dag_run.dag_run_note is None:
	dag_run.note = (new_note, current_user_id)
	else:
	dag_run.dag_run_note.content = new_note
	dag_run.dag_run_note.user_id = current_user_id
	session.commit()
	return dagrun_schema.dump(dag_run)

Solution:

-Use note paramater inside of first task so that after triggered dag it will update DagRun notes.
Example : PATCH-->BASE_URL/api/v1/dags/{dag_id}/dagRuns/{dag_run_id}/taskInstances/{task_id}/setNote

Alternative:

-Contribute new session to combine payload.