code: def get_dags(
      *,
      limit: int,
      offset: int = 0,
      tags: Collection[str] | None = None,
      dag_id_pattern: str | None = None,
      only_active: bool = True,
      paused: bool | None = None,
      order_by: str = "dag_id",
      fields: Collection[str] | None = None,
      session: Session = NEW_SESSION,
  ) -> APIResponse:
      """Get all DAGs."""
      allowed_attrs = ["dag_id"]
      dags_query = select(DagModel).where(~DagModel.is_subdag)
      if only_active:
          dags_query = dags_query.where(DagModel.is_active)
      if paused is not None:
          if paused:
              dags_query = dags_query.where(DagModel.is_paused)
          else:
              dags_query = dags_query.where(~DagModel.is_paused)
      if dag_id_pattern:
          dags_query = dags_query.where(
              DagModel.dag_id.ilike(f"%{dag_id_pattern}%")
          )
  
      readable_dags = get_auth_manager().get_permitted_dag_ids(
          user=g.user
      )
  
      dags_query = dags_query.where(DagModel.dag_id.in_(readable_dags))
      if tags:
          cond = [DagModel.tags.any(DagTag.name == tag) for tag in tags]
          dags_query = dags_query.where(or_(*cond))
  
      total_entries = get_query_count(dags_query, session=session)
      dags_query = apply_sorting(
          dags_query, order_by, {}, allowed_attrs
      )
      dags = session.scalars(
          dags_query.offset(offset).limit(limit)
      ).all()
  
      try:
          dags_collection_schema = (
              DAGCollectionSchema(
                  only=[f"dags.{field}" for field in fields]
              )
              if fields
              else DAGCollectionSchema()
          )
          return dags_collection_schema.dump(
              DAGCollection(dags=dags, total_entries=total_entries)
          )
      except ValueError as e:
          raise BadRequest("DAGCollectionSchema error", detail=str(e))
  
imports: from __future__ import annotations
  from http import HTTPStatus
  from typing import TYPE_CHECKING, Collection
  from connexion import NoContent
  from flask import g, request
  from marshmallow import ValidationError
  from sqlalchemy import select, update
  from sqlalchemy.sql.expression import or_
  from airflow.api_connexion import security
  from airflow.api_connexion.exceptions import AlreadyExists, BadRequest, NotFound
  from airflow.api_connexion.parameters import apply_sorting, check_limit, format_parameters
  from airflow.api_connexion.schemas.dag_schema import DAGCollection, DAGCollectionSchema, DAGDetailSchema, DAGSchema, dag_schema, dags_collection_schema
  from airflow.exceptions import AirflowException, DagNotFound
  from airflow.models.dag import DagModel, DagTag
  from airflow.utils.airflow_flask_app import get_airflow_app
  from airflow.utils.db import get_query_count
  from airflow.utils.session import NEW_SESSION, provide_session
  from airflow.www.extensions.init_auth_manager import get_auth_manager
label_desc: Use image node_id 3 for calling a global function with example usage: get_dags() and returns: dags_collection_schema