code: def read_database(  # noqa: D417
      query: str | Selectable,
      connection: ConnectionOrCursor | str,
      *,
      iter_batches: bool = False,
      batch_size: int | None = None,
      schema_overrides: SchemaDict | None = None,
      execute_options: dict[str, Any] | None = None,
      **kwargs: Any,
  ) -> DataFrame | Iterable[DataFrame]:
      """
      Read the results of a SQL query into a DataFrame, given a connection object.
  
      Parameters
      ----------
      query
          SQL query to execute (if using a SQLAlchemy connection object this can
          be a suitable "Selectable", otherwise it is expected to be a string).
      connection
          An instantiated connection (or cursor/client object) that the query can be
          executed against. Can also pass a valid ODBC connection string, starting with
          "Driver=", in which case the `arrow-odbc` package will be used to establish
          the connection and return Arrow-native data to Polars.
      iter_batches
          Return an iterator of DataFrames, where each DataFrame represents a batch of
          data returned by the query; this can be useful for processing large resultsets
          in a memory-efficient manner. If supported by the backend, this value is passed
          to the underlying query execution method (note that very low values will
          typically result in poor performance as it will result in many round-trips to
          the database as the data is returned). If the backend does not support changing
          the batch size then a single DataFrame is yielded from the iterator.
      batch_size
          Indicate the size of each batch when `iter_batches` is True (note that you can
          still set this when `iter_batches` is False, in which case the resulting
          DataFrame is constructed internally using batched return before being returned
          to you. Note that some backends may support batched operation but not allow for
          an explicit size; in this case you will still receive batches, but their exact
          size will be determined by the backend (so may not equal the value set here).
      schema_overrides
          A dictionary mapping column names to dtypes, used to override the schema
          inferred from the query cursor or given by the incoming Arrow data (depending
          on driver/backend). This can be useful if the given types can be more precisely
          defined (for example, if you know that a given column can be declared as `u32`
          instead of `i64`).
      execute_options
          These options will be passed through into the underlying query execution method
          as kwargs. In the case of connections made using an ODBC string (which use
          `arrow-odbc`) these options are passed to the `read_arrow_batches_from_odbc`
          method.
  
      Notes
      -----
      * This function supports a wide range of native database drivers (ranging from local
        databases such as SQLite to large cloud databases such as Snowflake), as well as
        generic libraries such as ADBC, SQLAlchemy and various flavours of ODBC. If the
        backend supports returning Arrow data directly then this facility will be used to
        efficiently instantiate the DataFrame; otherwise, the DataFrame is initialised
        from row-wise data.
  
      * Support for Arrow Flight SQL data is available via the `adbc-driver-flightsql`
        package; see https://arrow.apache.org/adbc/current/driver/flight_sql.html for
        more details about using this driver (notable databases implementing Flight SQL
        include Dremio and InfluxDB).
  
      * The `read_database_uri` function is likely to be noticeably faster than
        `read_database` if you are using a SQLAlchemy or DBAPI2 connection, as
        `connectorx` will optimise translation of the result set into Arrow format
        in Rust, whereas these libraries will return row-wise data to Python *before*
        we can load into Arrow. Note that you can easily determine the connection's
        URI from a SQLAlchemy engine object by calling `str(conn.engine.url)`.
  
      * If polars has to create a cursor from your connection in order to execute the
        query then that cursor will be automatically closed when the query completes;
        however, polars will *never* close any other connection or cursor.
  
      See Also
      --------
      read_database_uri : Create a DataFrame from a SQL query using a URI string.
  
      Examples
      --------
      Instantiate a DataFrame from a SQL query against a user-supplied connection:
  
      >>> df = pl.read_database(
      ...     query="SELECT * FROM test_data",
      ...     connection=user_conn,
      ...     schema_overrides={"normalised_score": pl.UInt8},
      ... )  # doctest: +SKIP
  
      Use a parameterised SQLAlchemy query, passing named values via `execute_options`:
  
      >>> df = pl.read_database(
      ...     query="SELECT * FROM test_data WHERE metric > :value",
      ...     connection=alchemy_conn,
      ...     execute_options={"parameters": {"value": 0}},
      ... )  # doctest: +SKIP
  
      Use 'qmark' style parameterisation; values are still passed via `execute_options`,
      but in this case the "parameters" value is a sequence of literals, not a dict:
  
      >>> df = pl.read_database(
      ...     query="SELECT * FROM test_data WHERE metric > ?",
      ...     connection=alchemy_conn,
      ...     execute_options={"parameters": [0]},
      ... )  # doctest: +SKIP
  
      Instantiate a DataFrame using an ODBC connection string (requires `arrow-odbc`)
      setting upper limits on the buffer size of variadic text/binary columns, returning
      the result as an iterator over DataFrames containing batches of 1000 rows:
  
      >>> for df in pl.read_database(
      ...     query="SELECT * FROM test_data",
      ...     connection="Driver={PostgreSQL};Server=localhost;Port=5432;Database=test;Uid=usr;Pwd=",
      ...     execute_options={"max_text_size": 512, "max_binary_size": 1024},
      ...     iter_batches=True,
      ...     batch_size=1000,
      ... ):
      ...     do_something(df)  # doctest: +SKIP
  
      """  # noqa: W505
      if isinstance(connection, str):
          # check for odbc connection string
          if (
              re.sub(r"\s", "", connection[:20])
              .lower()
              .startswith("driver=")
          ):
              try:
                  import arrow_odbc  # noqa: F401
              except ModuleNotFoundError:
                  raise ModuleNotFoundError(
                      "use of an ODBC connection string requires the `arrow-odbc` package"
                      "\n\nPlease run: pip install arrow-odbc"
                  ) from None
  
              connection = ODBCCursorProxy(connection)
          else:
              # otherwise looks like a call to read_database_uri
              issue_deprecation_warning(
                  message="Use of a string URI with 'read_database' is deprecated; use `read_database_uri` instead",
                  version="0.19.0",
              )
              if iter_batches or batch_size:
                  raise InvalidOperationError(
                      "Batch parameters are not supported for `read_database_uri`"
                  )
              if not isinstance(query, (list, str)):
                  raise TypeError(
                      f"`read_database_uri` expects one or more string queries; found {type(query)}"
                  )
              return read_database_uri(
                  query,
                  uri=connection,
                  schema_overrides=schema_overrides,
                  **kwargs,
              )
  
      # note: can remove this check (and **kwargs) once we drop the
      # pass-through deprecation support for read_database_uri
      if kwargs:
          raise ValueError(
              f"`read_database` **kwargs only exist for passthrough to `read_database_uri`: found {kwargs!r}"
          )
  
      # return frame from arbitrary connections using the executor abstraction
      with ConnectionExecutor(connection) as cx:
          return cx.execute(
              query=query,
              options=execute_options,
          ).to_polars(
              batch_size=batch_size,
              iter_batches=iter_batches,
              schema_overrides=schema_overrides,
          )
  
imports: from __future__ import annotations
  import re
  import sys
  from importlib import import_module
  from inspect import Parameter, signature
  from typing import TYPE_CHECKING, Any, Iterable, Literal, Sequence, TypedDict, overload
  from polars.convert import from_arrow
  from polars.exceptions import InvalidOperationError, UnsuitableSQLError
  from polars.utils.deprecation import issue_deprecation_warning
label_desc: Use image node_id 3 for calling a global function with example usage: read_database(query, connection) and returns: cx, read_database_uri