Database Module

database

Database connection and session management.

This module provides async database engine initialization and session management for SQLModel/SQLAlchemy. It uses connection pooling and is designed to be initialized once during application startup.

Functions

init_engine

init_engine()

Initialize the async engine and sessionmaker.

Creates a connection pool and session factory. Should be called once during application startup (typically in FastAPI lifespan). Reads DATABASE_URL from environment variables.

Raises:

Type	Description
RuntimeError	If DATABASE_URL environment variable is not set.

Note

Uses connection pooling with: - pool_size: 5 connections - max_overflow: 10 additional connections - pool_pre_ping: True (validates connections before use)

Source code in bitonicai/database.py

def init_engine() -> None:
    """Initialize the async engine and sessionmaker.

    Creates a connection pool and session factory. Should be called once during
    application startup (typically in FastAPI lifespan). Reads DATABASE_URL
    from environment variables.

    Raises:
        RuntimeError: If DATABASE_URL environment variable is not set.

    Note:
        Uses connection pooling with:
        - pool_size: 5 connections
        - max_overflow: 10 additional connections
        - pool_pre_ping: True (validates connections before use)
    """
    global engine, async_session

    if engine is not None and async_session is not None:
        return

    database_url = os.getenv("DATABASE_URL")
    if not database_url:
        raise RuntimeError("DATABASE_URL environment variable is not set.")

    engine = create_async_engine(
        database_url,
        echo=False,
        pool_size=5,
        max_overflow=10,
        pool_pre_ping=True,
    )
    async_session = async_sessionmaker[AsyncSession](
        engine,
        expire_on_commit=False,
        class_=AsyncSession,
    )

get_session async

get_session()

FastAPI dependency that provides a scoped async session.

Yields a database session that is automatically closed after the request. Use this as a FastAPI dependency in route handlers.

Yields:

Name	Type	Description
AsyncSession	AsyncGenerator[AsyncSession, None]	Database session instance.

Example

@app.get("/items")
async def get_items(db: AsyncSession = Depends(get_session)):
    result = await db.exec(select(Item))
    return result.all()

Source code in bitonicai/database.py

async def get_session() -> AsyncGenerator[AsyncSession, None]:
    """FastAPI dependency that provides a scoped async session.

    Yields a database session that is automatically closed after the request.
    Use this as a FastAPI dependency in route handlers.

    Yields:
        AsyncSession: Database session instance.

    Example:
        ```python
        @app.get("/items")
        async def get_items(db: AsyncSession = Depends(get_session)):
            result = await db.exec(select(Item))
            return result.all()
        ```
    """
    if async_session is None:
        # Fallback initialization in case startup was skipped.
        init_engine()

    assert async_session is not None  # for type checkers
    async with async_session() as session:
        yield session

create_all_tables async

create_all_tables()

Create database tables using SQLModel metadata.

Creates all tables defined in SQLModel entities. This is typically only used for development or testing. In production, use Alembic migrations instead.

Raises:

Type	Description
RuntimeError	If engine is not initialized (call init_engine first).

Source code in bitonicai/database.py

async def create_all_tables() -> None:
    """Create database tables using SQLModel metadata.

    Creates all tables defined in SQLModel entities. This is typically only
    used for development or testing. In production, use Alembic migrations instead.

    Raises:
        RuntimeError: If engine is not initialized (call init_engine first).
    """
    if engine is None:
        raise RuntimeError("Engine not initialized. Call init_engine first.")

    async with engine.begin() as conn:
        await conn.run_sync(SQLModel.metadata.create_all)

dispose_engine async

dispose_engine()

Dispose of the engine and close all connections.

Closes all database connections and cleans up the connection pool. Should be called during application shutdown (typically in FastAPI lifespan).

Source code in bitonicai/database.py

async def dispose_engine() -> None:
    """Dispose of the engine and close all connections.

    Closes all database connections and cleans up the connection pool.
    Should be called during application shutdown (typically in FastAPI lifespan).
    """
    global engine, async_session

    if engine is not None:
        await engine.dispose()

    engine = None
    async_session = None