Repository: apache/airflow
File: airflow-core/src/airflow/api/common/trigger_dag.py (Python)

Code before:
        conf=run_conf,
        run_type=DagRunType.MANUAL,
        triggered_by=triggered_by,
        triggering_user_name=triggering_user_name,
        note=note,
        state=DagRunState.QUEUED,
        session=session,
    )

    return dag_run


@provide_session
def trigger_dag(
    dag_id: str,
    *,
    triggered_by: DagRunTriggeredByType,
    triggering_user_name: str | None = None,
    run_after: datetime | None = None,
    run_id: str | None = None,
    conf: dict | str | None = None,
    logical_date: datetime | None = None,
    replace_microseconds: bool = True,
    note: str | None = None,
    session: Session = NEW_SESSION,
) -> DagRun | None:
    """
    Triggers execution of DAG specified by dag_id.

    :param dag_id: DAG ID
    :param triggered_by: the entity which triggers the dag_run
    :param triggering_user_name: the user name who triggers the dag_run
    :param run_after: the datetime before which dag won't run
    :param run_id: ID of the dag_run
    :param conf: configuration
    :param logical_date: date of execution
    :param replace_microseconds: whether microseconds should be zeroed
    :param session: Unused. Only added in compatibility with database isolation mode
    :return: first dag run triggered - even if more than one Dag Runs were triggered or None
    """
    dag_model = DagModel.get_current(dag_id, session=session)
    if dag_model is None:
        raise DagNotFound(f"Dag id {dag_id} not found in DagModel")

    dagbag = DBDagBag()
    dr = _trigger_dag(
        dag_id=dag_id,
        dag_bag=dagbag,
        run_id=run_id,
        run_after=run_after or timezone.utcnow(),
        conf=conf,

Review comment (suggestion):
To be sure for backward compatibility we need a cadwyn migration update.

Can you add that to the `/api_fastapi/execution_api/versions/v2026_03_31.py` version, this is the 3.2 migration file.

Code after:
        run_after=run_after,
        conf=run_conf,
        run_type=DagRunType.MANUAL,
        triggered_by=triggered_by,
        triggering_user_name=triggering_user_name,
        note=note,
        state=DagRunState.QUEUED,
        partition_key=partition_key,
        session=session,
    )

    return dag_run


@provide_session
def trigger_dag(
    dag_id: str,
    *,
    triggered_by: DagRunTriggeredByType,
    triggering_user_name: str | None = None,
    run_after: datetime | None = None,
    run_id: str | None = None,
    conf: dict | str | None = None,
    logical_date: datetime | None = None,
    replace_microseconds: bool = True,
    note: str | None = None,
    partition_key: str | None = None,
    session: Session = NEW_SESSION,
) -> DagRun | None:
    """
    Triggers execution of DAG specified by dag_id.

    :param dag_id: DAG ID
    :param triggered_by: the entity which triggers the dag_run
    :param triggering_user_name: the user name who triggers the dag_run
    :param run_after: the datetime before which dag won't run
    :param run_id: ID of the dag_run
    :param conf: configuration
    :param logical_date: date of execution
    :param replace_microseconds: whether microseconds should be zeroed
    :param session: Unused. Only added in compatibility with database isolation mode
    :return: first dag run triggered - even if more than one Dag Runs were triggered or None
    """
    dag_model = DagModel.get_current(dag_id, session=session)
    if dag_model is None:
        raise DagNotFound(f"Dag id {dag_id} not found in DagModel")

    dagbag = DBDagBag()
    dr = _trigger_dag(
        dag_id=dag_id,
        dag_bag=dagbag,