flip_api.db.database

Attributes

_POOL_RECYCLE_SECONDS
_rds_client
_rds_client_lock
_engine
_engine_lock

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_engine(→ sqlalchemy.engine.Engine)	Return the shared SQLAlchemy engine, building it on first use.
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: sqlalchemy.engine.Engine | None = None

flip_api.db.database._engine_lock

flip_api.db.database.get_engine() → sqlalchemy.engine.Engine

Return the shared SQLAlchemy engine, building it on first use.

Returns:

The lazily-built, module-cached engine for the active environment.

Return type:

Engine

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

Create a new SQLModel session.

The with block is load-bearing: FastAPI finalises this dependency by throwing the endpoint’s exception into the generator at the yield (or closing it on disconnect/cancellation), so a bare yield followed by session.close() never closes the session on those paths — the checked-out connection is stranded idle in transaction until the process restarts, and enough of them exhaust the pool and take the whole API down (FLIP#773).

Yields:

Session – A new SQLModel session.