Initialize

SessionDatabaseUnitOfWorkType = TypeVar('SessionDatabaseUnitOfWorkType', bound=SessionDatabaseUnitOfWork) module-attribute

SessionMakerType = async_sessionmaker | Callable[[], AsyncSession] module-attribute

logger = logging.getLogger(__name__) module-attribute

backend_redis_uri_from_settings()

For backend exclusive use (not reflex).

Reflex uses /0 (for some reason it is much slower if using /1)

db_uri_from_settings(asynchronous=True, driver='postgresql', async_addition='+asyncpg', user=settings.POSTGRES_USER, password=settings.POSTGRES_PASSWORD, host=settings.POSTGRES_HOST, port=settings.POSTGRES_PORT, db=settings.POSTGRES_DB)

get_session() async

Get a new session from the db.

get_sessionmaker()

Get the session maker.

initialize_db(database_uri)

Initialize the db engine and session maker.

initialize_sqlite_tables(engine)

Initialize tables in database.

Note: This should only be done for temporary sqlite databases, the real database should be fully managed via alembic migrations.

make_db_uow_generator(uow_class, session_maker=None)

Generate a factory function for a database unit of work.

Lazily uses the session_maker to create a new UOW when needed.

reset_db_engine_and_session()

Un-initialize the db engine and session maker.

set_db_engine_and_session(engine, session_maker)

Set the db engine and session maker.