code: def get_dag_runs(
      *,
      dag_id: str,
      start_date_gte: str | None = None,
      start_date_lte: str | None = None,
      execution_date_gte: str | None = None,
      execution_date_lte: str | None = None,
      end_date_gte: str | None = None,
      end_date_lte: str | None = None,
      updated_at_gte: str | None = None,
      updated_at_lte: str | None = None,
      state: list[str] | None = None,
      offset: int | None = None,
      limit: int | None = None,
      order_by: str = "id",
      fields: Collection[str] | None = None,
      session: Session = NEW_SESSION,
  ):
      """Get all DAG Runs."""
      query = select(DagRun)
  
      #  This endpoint allows specifying ~ as the dag_id to retrieve DAG Runs for all DAGs.
      if dag_id == "~":
          query = query.where(
              DagRun.dag_id.in_(
                  get_auth_manager().get_permitted_dag_ids(
                      methods=["GET"], user=g.user
                  )
              )
          )
      else:
          query = query.where(DagRun.dag_id == dag_id)
  
      if state:
          query = query.where(DagRun.state.in_(state))
  
      dag_run, total_entries = _fetch_dag_runs(
          query,
          end_date_gte=end_date_gte,
          end_date_lte=end_date_lte,
          execution_date_gte=execution_date_gte,
          execution_date_lte=execution_date_lte,
          start_date_gte=start_date_gte,
          start_date_lte=start_date_lte,
          updated_at_gte=updated_at_gte,
          updated_at_lte=updated_at_lte,
          limit=limit,
          offset=offset,
          order_by=order_by,
          session=session,
      )
      try:
          dagrun_collection_schema = (
              DAGRunCollectionSchema(context={"fields": fields})
              if fields
              else DAGRunCollectionSchema()
          )
          return dagrun_collection_schema.dump(
              DAGRunCollection(
                  dag_runs=dag_run, total_entries=total_entries
              )
          )
      except ValueError as e:
          raise BadRequest(
              "DAGRunCollectionSchema error", detail=str(e)
          )
  
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 5 for calling a global function with example usage: get_dag_runs() and returns: dagrun_collection_schema