Database Dependencies ¶

Setup with Settings and Environment ¶

from typing import Annotated, AsyncGenerator
from pathlib import Path

from pydantic_settings import BaseSettings, SettingsConfigDict
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker
from contextlib import asynccontextmanager
from typing_extensions import Self

# Load settings from .env file and/or from environment variables
class DatabaseSettings(BaseSettings):
    database_url: str = "sqlite+aiosqlite:///database.db"

    @classmethod
    def from_env(cls) -> Self:
        return cls()

def get_db_settings() -> DatabaseSettings:
    return DatabaseSettings()


## Create AsyncSessionMaker

def get_sessionmaker(
    settings: Annotated[DatabaseSettings, DatabaseSettings.from_env, "singleton"],
) -> async_sessionmaker[AsyncSession]:
    """
    Create an async sessionmaker from database settings.
    This is typically a singleton in your app.
    """
    engine = create_async_engine(
        settings.database_url,
        echo=False,  # Set to True for SQL debugging
        future=True,
    )

    return async_sessionmaker(
        bind=engine,
        class_=AsyncSession,
        autocommit=False,
        autoflush=False,
        expire_on_commit=False,
    )

# We can share sessionmaker between dependencies
type SessionMaker = Annotated[async_sessionmaker[AsyncSession], get_sessionmaker, "singleton"]

# Create a session for each request

@asynccontextmanager
async def get_session(sessionmaker: SessionMaker) -> AsyncGenerator[AsyncSession, None]:
    """
    Get a new database async session with automatic commit/rollback.
    Yields the session and handles cleanup.
    """
    async with sessionmaker() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise
        finally:
            await session.close()

# For concurrenty safety we will create a new session for each dependency
GetSession = Annotated[AsyncSession, get_session, "transient"]

Usage in Handlers ¶

from stario import Query
from sqlalchemy import select

@app.query("/users")
async def list_users(db: GetSession):
    """List all users from database."""
    stmt = select(User)
    result = await db.execute(stmt)
    users = result.scalars().all()
    return div([render_user(u) for u in users])

Key Points ¶

• Settings: Use BaseSettings to load database URLs from environment
• SessionMaker: Create once (singleton), reuse for all requests
• Sessions: Create transiently, auto-commit on success, auto-rollback on error
• Dependency Injection: Use Annotated for clean, reusable dependency injection
• Type Aliases: Define custom type aliases for frequently-used dependencies to reduce boilerplate
• Error Handling: Sessions automatically handle commit/rollback in the context manager