flip_api.db.database
====================

.. py:module:: flip_api.db.database


Attributes
----------

.. autoapisummary::

   flip_api.db.database._POOL_RECYCLE_SECONDS
   flip_api.db.database._rds_client
   flip_api.db.database._rds_client_lock
   flip_api.db.database.engine


Functions
---------

.. autoapisummary::

   flip_api.db.database._get_rds_client
   flip_api.db.database._generate_db_auth_token
   flip_api.db.database._do_connect_listener
   flip_api.db.database._build_engine
   flip_api.db.database.get_session


Module Contents
---------------

.. py:data:: _POOL_RECYCLE_SECONDS
   :value: 1500


.. py:data:: _rds_client
   :type:  Any
   :value: None


.. py:data:: _rds_client_lock

.. py:function:: _get_rds_client() -> Any

   Return a cached boto3 RDS client for the configured region.


.. py:function:: _generate_db_auth_token() -> str

   Mint a short-lived (~15 min) IAM auth token to use as the DB password.

   ``generate_db_auth_token`` is a local SigV4 signing operation (no network
   round-trip), so it is cheap to call on the connection path. A fresh token
   is produced for every new physical connection, so token expiry and RDS
   secret rotation are both handled transparently — the app holds no static
   DB credential.

   :returns: IAM authentication token for the configured DB user and host.
   :rtype: str

   :raises Exception: re-raises any error from minting the token (after logging it),
       so the failure still surfaces loudly on the connection path.


.. py:function:: _do_connect_listener(_dialect: object, _conn_rec: object, _cargs: object, cparams: dict[str, Any]) -> None

   SQLAlchemy ``do_connect`` hook: inject a freshly-minted IAM token as the password.

   The token is passed as a connection parameter (not embedded in the engine
   URL), so it never needs URL-encoding and is regenerated on each new
   physical connection.


.. py:function:: _build_engine() -> sqlalchemy.engine.Engine

   Build the SQLAlchemy engine for the active environment.

   Production authenticates to Postgres through RDS Proxy with a per-connection
   IAM token (passwordless URL + a ``do_connect`` hook), so the app holds no
   static credential and RDS secret rotation is a non-event (FLIP#556). Dev
   uses the static ``POSTGRES_PASSWORD`` from the environment.


.. py:data:: engine

.. py:function:: get_session() -> collections.abc.Generator[sqlmodel.Session, None, None]

   Create a new SQLModel session.

   :returns: A new SQLModel session.
   :rtype: Session