code: def post_dag_run(
      *, dag_id: str, session: Session = NEW_SESSION
  ) -> APIResponse:
      """Trigger a DAG."""
      dm = session.scalar(
          select(DagModel)
          .where(DagModel.is_active, DagModel.dag_id == dag_id)
          .limit(1)
      )
      if not dm:
          raise NotFound(
              title="DAG not found",
              detail=f"DAG with dag_id: '{dag_id}' not found",
          )
      if dm.has_import_errors:
          raise BadRequest(
              title="DAG cannot be triggered",
              detail=f"DAG with dag_id: '{dag_id}' has import errors",
          )
      try:
          post_body = dagrun_schema.load(
              get_json_request_dict(), session=session
          )
      except ValidationError as err:
          raise BadRequest(detail=str(err))
  
      logical_date = pendulum.instance(post_body["execution_date"])
      run_id = post_body["run_id"]
      dagrun_instance = session.scalar(
          select(DagRun)
          .where(
              DagRun.dag_id == dag_id,
              or_(
                  DagRun.run_id == run_id,
                  DagRun.execution_date == logical_date,
              ),
          )
          .limit(1)
      )
      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,
              )
              dag_run_note = post_body.get("note")
              if dag_run_note:
                  current_user_id = get_auth_manager().get_user_id()
                  dag_run.note = (dag_run_note, current_user_id)
              return dagrun_schema.dump(dag_run)
          except ValueError as ve:
              raise BadRequest(detail=str(ve))
  
      if dagrun_instance.execution_date == logical_date:
          raise AlreadyExists(
              detail=(
                  f"DAGRun with DAG ID: '{dag_id}' and "
                  f"DAGRun logical date: '{logical_date.isoformat(sep=' ')}' already exists"
              ),
          )
  
      raise AlreadyExists(
          detail=f"DAGRun with DAG ID: '{dag_id}' and DAGRun ID: '{run_id}' already exists"
      )
  
imports: from __future__ import annotations
  from http import HTTPStatus
  from typing import TYPE_CHECKING, Collection
  import pendulum
  from connexion import NoContent
  from flask import g
  from marshmallow import ValidationError
  from sqlalchemy import delete, or_, select
  from airflow.api.common.mark_tasks import set_dag_run_state_to_failed, set_dag_run_state_to_queued, set_dag_run_state_to_success
  from airflow.api_connexion import security
  from airflow.api_connexion.endpoints.request_dict import get_json_request_dict
  from airflow.api_connexion.exceptions import AlreadyExists, BadRequest, NotFound
  from airflow.api_connexion.parameters import apply_sorting, check_limit, format_datetime, format_parameters
  from airflow.api_connexion.schemas.dag_run_schema import DAGRunCollection, DAGRunCollectionSchema, DAGRunSchema, clear_dagrun_form_schema, dagrun_collection_schema, dagrun_schema, dagruns_batch_form_schema, set_dagrun_note_form_schema, set_dagrun_state_form_schema
  from airflow.api_connexion.schemas.dataset_schema import DatasetEventCollection, dataset_event_collection_schema
  from airflow.api_connexion.schemas.task_instance_schema import TaskInstanceReferenceCollection, task_instance_reference_collection_schema
  from airflow.auth.managers.models.resource_details import DagAccessEntity
  from airflow.models import DagModel, DagRun
  from airflow.security import permissions
  from airflow.utils.airflow_flask_app import get_airflow_app
  from airflow.utils.db import get_query_count
  from airflow.utils.log.action_logger import action_event_from_permission
  from airflow.utils.session import NEW_SESSION, provide_session
  from airflow.utils.state import DagRunState
  from airflow.utils.types import DagRunType
  from airflow.www.decorators import action_logging
  from airflow.www.extensions.init_auth_manager import get_auth_manager
label_desc: Use image node_id 7 for calling a global function with example usage: post_dag_run() and returns: dagrun_schema