SQLModel Integration

Database session management for Xitzin applications using SQLModel.

Optional Dependency

Requires the sqlmodel extra: pip install xitzin[sqlmodel]

Database Migrations

For schema evolution in production, see the Database Migrations guide for Alembic integration.

SessionMiddleware

Creates middleware that manages database sessions per request.

SessionMiddleware

SessionMiddleware(
    engine: Engine, *, autoflush: bool = True
) -> Callable[
    [
        "Request",
        Callable[["Request"], Awaitable["GeminiResponse"]],
    ],
    Awaitable["GeminiResponse"],
]

Create a middleware that manages database sessions per request.

The session is stored in request.state.db and automatically committed on success or rolled back on error.

Parameters:

Name	Type	Description	Default
engine	Engine	SQLAlchemy engine instance.	required
autoflush	bool	If True, flush before queries. Defaults to True.	True

Returns:

Type	Description
Callable[['Request', Callable[['Request'], Awaitable['GeminiResponse']]], Awaitable['GeminiResponse']]	Middleware function compatible with @app.middleware.

Example

engine = create_engine("sqlite:///./database.db") app.middleware(SessionMiddleware(engine))

Source code in src/xitzin/sqlmodel.py

def SessionMiddleware(
    engine: Engine,
    *,
    autoflush: bool = True,
) -> Callable[
    ["Request", Callable[["Request"], Awaitable["GeminiResponse"]]],
    Awaitable["GeminiResponse"],
]:
    """Create a middleware that manages database sessions per request.

    The session is stored in request.state.db and automatically committed
    on success or rolled back on error.

    Args:
        engine: SQLAlchemy engine instance.
        autoflush: If True, flush before queries. Defaults to True.

    Returns:
        Middleware function compatible with @app.middleware.

    Example:
        engine = create_engine("sqlite:///./database.db")
        app.middleware(SessionMiddleware(engine))
    """

    async def middleware(
        request: "Request",
        call_next: Callable[["Request"], Awaitable["GeminiResponse"]],
    ) -> "GeminiResponse":
        session = Session(engine, autoflush=autoflush)
        request.state.db = session

        try:
            response = await call_next(request)
            session.commit()
            return response
        except Exception:
            session.rollback()
            raise
        finally:
            session.close()

    return middleware

get_session

Retrieves the database session from the current request.

get_session

get_session(request: 'Request') -> Session

Get the database session from the current request.

This helper retrieves the session created by SessionMiddleware.

Parameters:

Name	Type	Description	Default
request	'Request'	The current Xitzin request object.	required

Returns:

Type	Description
Session	SQLModel Session instance.

Raises:

Type	Description
AttributeError	If SessionMiddleware is not configured.

Example

@app.gemini("/users/{user_id}") def get_user(request: Request, user_id: int): session = get_session(request) user = session.get(User, user_id) return f"# {user.name}"

Source code in src/xitzin/sqlmodel.py

def get_session(request: "Request") -> Session:
    """Get the database session from the current request.

    This helper retrieves the session created by SessionMiddleware.

    Args:
        request: The current Xitzin request object.

    Returns:
        SQLModel Session instance.

    Raises:
        AttributeError: If SessionMiddleware is not configured.

    Example:
        @app.gemini("/users/{user_id}")
        def get_user(request: Request, user_id: int):
            session = get_session(request)
            user = session.get(User, user_id)
            return f"# {user.name}"
    """
    if not hasattr(request.state, "db"):
        raise AttributeError(
            "No database session found. Did you add SessionMiddleware?"
        )
    return request.state.db

init_db

Registers database lifecycle hooks for table creation and cleanup.

init_db

init_db(
    app: "Xitzin",
    engine: Engine,
    *,
    create_tables: bool = True,
    drop_all: bool = False,
) -> None

Initialize database with lifecycle hooks.

This helper registers startup/shutdown hooks to manage table creation and engine cleanup.

Parameters:

Name	Type	Description	Default
app	'Xitzin'	Xitzin application instance.	required
engine	Engine	SQLAlchemy engine instance.	required
create_tables	bool	Create all tables on startup. Defaults to True.	True
drop_all	bool	Drop all tables before creating. Defaults to False.	False

Warning

Setting drop_all=True will DELETE ALL DATA on startup!

Example

app = Xitzin() engine = create_engine("sqlite:///./database.db") init_db(app, engine)

Source code in src/xitzin/sqlmodel.py

def init_db(
    app: "Xitzin",
    engine: Engine,
    *,
    create_tables: bool = True,
    drop_all: bool = False,
) -> None:
    """Initialize database with lifecycle hooks.

    This helper registers startup/shutdown hooks to manage table creation
    and engine cleanup.

    Args:
        app: Xitzin application instance.
        engine: SQLAlchemy engine instance.
        create_tables: Create all tables on startup. Defaults to True.
        drop_all: Drop all tables before creating. Defaults to False.

    Warning:
        Setting drop_all=True will DELETE ALL DATA on startup!

    Example:
        app = Xitzin()
        engine = create_engine("sqlite:///./database.db")
        init_db(app, engine)
    """

    @app.on_startup
    def create_db_tables() -> None:
        if drop_all:
            SQLModel.metadata.drop_all(engine)
        if create_tables:
            SQLModel.metadata.create_all(engine)

    @app.on_shutdown
    def dispose_engine() -> None:
        engine.dispose()

Re-exports

The following are re-exported from sqlmodel for convenience:

• SQLModel - Base class for models
• Session - Database session class
• create_engine - Engine factory function