code: def prepare_engine_args(
      disable_connection_pool=False, pool_class=None
  ):
      """Prepare SQLAlchemy engine args."""
      default_args = {}
      for dialect, default in DEFAULT_ENGINE_ARGS.items():
          if SQL_ALCHEMY_CONN.startswith(dialect):
              default_args = default.copy()
              break
  
      engine_args: dict = conf.getjson("database", "sql_alchemy_engine_args", fallback=default_args)  # type: ignore
  
      if pool_class:
          # Don't use separate settings for size etc, only those from sql_alchemy_engine_args
          engine_args["poolclass"] = pool_class
      elif disable_connection_pool or not conf.getboolean(
          "database", "SQL_ALCHEMY_POOL_ENABLED"
      ):
          engine_args["poolclass"] = NullPool
          log.debug("settings.prepare_engine_args(): Using NullPool")
      elif not SQL_ALCHEMY_CONN.startswith("sqlite"):
          # Pool size engine args not supported by sqlite.
          # If no config value is defined for the pool size, select a reasonable value.
          # 0 means no limit, which could lead to exceeding the Database connection limit.
          pool_size = conf.getint(
              "database", "SQL_ALCHEMY_POOL_SIZE", fallback=5
          )
  
          # The maximum overflow size of the pool.
          # When the number of checked-out connections reaches the size set in pool_size,
          # additional connections will be returned up to this limit.
          # When those additional connections are returned to the pool, they are disconnected and discarded.
          # It follows then that the total number of simultaneous connections
          # the pool will allow is pool_size + max_overflow,
          # and the total number of "sleeping" connections the pool will allow is pool_size.
          # max_overflow can be set to -1 to indicate no overflow limit;
          # no limit will be placed on the total number
          # of concurrent connections. Defaults to 10.
          max_overflow = conf.getint(
              "database", "SQL_ALCHEMY_MAX_OVERFLOW", fallback=10
          )
  
          # The DB server already has a value for wait_timeout (number of seconds after
          # which an idle sleeping connection should be killed). Since other DBs may
          # co-exist on the same server, SQLAlchemy should set its
          # pool_recycle to an equal or smaller value.
          pool_recycle = conf.getint(
              "database", "SQL_ALCHEMY_POOL_RECYCLE", fallback=1800
          )
  
          # Check connection at the start of each connection pool checkout.
          # Typically, this is a simple statement like "SELECT 1", but may also make use
          # of some DBAPI-specific method to test the connection for liveness.
          # More information here:
          # https://docs.sqlalchemy.org/en/14/core/pooling.html#disconnect-handling-pessimistic
          pool_pre_ping = conf.getboolean(
              "database", "SQL_ALCHEMY_POOL_PRE_PING", fallback=True
          )
  
          log.debug(
              "settings.prepare_engine_args(): Using pool settings. pool_size=%d, max_overflow=%d, "
              "pool_recycle=%d, pid=%d",
              pool_size,
              max_overflow,
              pool_recycle,
              os.getpid(),
          )
          engine_args["pool_size"] = pool_size
          engine_args["pool_recycle"] = pool_recycle
          engine_args["pool_pre_ping"] = pool_pre_ping
          engine_args["max_overflow"] = max_overflow
  
      # The default isolation level for MySQL (REPEATABLE READ) can introduce inconsistencies when
      # running multiple schedulers, as repeated queries on the same session may read from stale snapshots.
      # 'READ COMMITTED' is the default value for PostgreSQL.
      # More information here:
      # https://dev.mysql.com/doc/refman/8.0/en/innodb-transaction-isolation-levels.html"
  
      if SQL_ALCHEMY_CONN.startswith("mysql"):
          engine_args["isolation_level"] = "READ COMMITTED"
  
      # Allow the user to specify an encoding for their DB otherwise default
      # to utf-8 so jobs & users with non-latin1 characters can still use us.
      engine_args["encoding"] = conf.get(
          "database", "SQL_ENGINE_ENCODING", fallback="utf-8"
      )
  
      return engine_args
  
imports: from __future__ import annotations
  import atexit
  import functools
  import json
  import logging
  import os
  import sys
  import warnings
  from typing import TYPE_CHECKING, Any, Callable
  import pluggy
  from sqlalchemy import create_engine, exc, text
  from sqlalchemy.orm import scoped_session, sessionmaker
  from sqlalchemy.pool import NullPool
  from airflow import policies
  from airflow.configuration import AIRFLOW_HOME, WEBSERVER_CONFIG, conf
  from airflow.exceptions import RemovedInAirflow3Warning
  from airflow.executors import executor_constants
  from airflow.logging_config import configure_logging
  from airflow.utils.orm_event_handlers import setup_event_handlers
  from airflow.utils.state import State
  from airflow.utils.timezone import local_timezone, parse_timezone, utc
label_desc: Use image node_id 14 for calling a global function with example usage: prepare_engine_args(disable_connection_pool, pool_class) and returns: engine_args