code: def _fetch_dag_runs(
      query: Select,
      *,
      end_date_gte: str | None,
      end_date_lte: str | None,
      execution_date_gte: str | None,
      execution_date_lte: str | None,
      start_date_gte: str | None,
      start_date_lte: str | None,
      updated_at_gte: str | None = None,
      updated_at_lte: str | None = None,
      limit: int | None,
      offset: int | None,
      order_by: str,
      session: Session,
  ) -> tuple[list[DagRun], int]:
      if start_date_gte:
          query = query.where(DagRun.start_date >= start_date_gte)
      if start_date_lte:
          query = query.where(DagRun.start_date <= start_date_lte)
      # filter execution date
      if execution_date_gte:
          query = query.where(
              DagRun.execution_date >= execution_date_gte
          )
      if execution_date_lte:
          query = query.where(
              DagRun.execution_date <= execution_date_lte
          )
      # filter end date
      if end_date_gte:
          query = query.where(DagRun.end_date >= end_date_gte)
      if end_date_lte:
          query = query.where(DagRun.end_date <= end_date_lte)
      # filter updated at
      if updated_at_gte:
          query = query.where(DagRun.updated_at >= updated_at_gte)
      if updated_at_lte:
          query = query.where(DagRun.updated_at <= updated_at_lte)
  
      total_entries = get_query_count(query, session=session)
      to_replace = {"dag_run_id": "run_id"}
      allowed_filter_attrs = [
          "id",
          "state",
          "dag_id",
          "execution_date",
          "dag_run_id",
          "start_date",
          "end_date",
          "updated_at",
          "external_trigger",
          "conf",
      ]
      query = apply_sorting(
          query, order_by, to_replace, allowed_filter_attrs
      )
      return (
          session.scalars(query.offset(offset).limit(limit)).all(),
          total_entries,
      )
  
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 4 for calling a global function with example usage: _fetch_dag_runs(query) and returns: session,  total_entries