flip_api.db.database

Attributes

`_POOL_RECYCLE_SECONDS`
`_rds_client`
`_rds_client_lock`
`engine`

Functions

`_get_rds_client`(→ Any)	Return a cached boto3 RDS client for the configured region.
`_generate_db_auth_token`(→ str)	Mint a short-lived (~15 min) IAM auth token to use as the DB password.
`_do_connect_listener`(→ None)	SQLAlchemy `do_connect` hook: inject a freshly-minted IAM token as the password.
`_build_engine`(→ sqlalchemy.engine.Engine)	Build the SQLAlchemy engine for the active environment.
`get_session`(...)	Create a new SQLModel session.

Module Contents

flip_api.db.database._POOL_RECYCLE_SECONDS = 1500

flip_api.db.database._rds_client: Any = None

flip_api.db.database._rds_client_lock

flip_api.db.database._get_rds_client() → Any: Return a cached boto3 RDS client for the configured region.

flip_api.db.database._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.
Return type:: str
Raises:: Exception – re-raises any error from minting the token (after logging it), so the failure still surfaces loudly on the connection path.

flip_api.db.database._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.

flip_api.db.database._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.

flip_api.db.database.engine

flip_api.db.database.get_session() → collections.abc.Generator[sqlmodel.Session, None, None]

Create a new SQLModel session.

Returns:: A new SQLModel session.
Return type:: Session