code: def trigger_dag(
      dag_id: str,
      run_id: str | None = None,
      conf: dict | str | None = None,
      execution_date: datetime | None = None,
      replace_microseconds: bool = True,
  ) -> DagRun | None:
      """Triggers execution of DAG specified by dag_id.
  
      :param dag_id: DAG ID
      :param run_id: ID of the dag_run
      :param conf: configuration
      :param execution_date: date of execution
      :param replace_microseconds: whether microseconds should be zeroed
      :return: first dag run triggered - even if more than one Dag Runs were triggered or None
      """
      dag_model = DagModel.get_current(dag_id)
      if dag_model is None:
          raise DagNotFound(f"Dag id {dag_id} not found in DagModel")
  
      dagbag = DagBag(
          dag_folder=dag_model.fileloc, read_dags_from_db=True
      )
      triggers = _trigger_dag(
          dag_id=dag_id,
          dag_bag=dagbag,
          run_id=run_id,
          conf=conf,
          execution_date=execution_date,
          replace_microseconds=replace_microseconds,
      )
  
      return triggers[0] if triggers else None
  
file_path: airflow/airflow/api/common/trigger_dag.py
imports: from __future__ import annotations
  import json
  from typing import TYPE_CHECKING
  from airflow.exceptions import DagNotFound, DagRunAlreadyExists
  from airflow.models import DagBag, DagModel, DagRun
  from airflow.utils import timezone
  from airflow.utils.state import DagRunState
  from airflow.utils.types import DagRunType
label_desc: Use image node_id 2 for calling a global function with example usage: trigger_dag(dag_id, run_id, conf, execution_date, replace_microseconds) and returns: unknown