Database & ORM

axion ships optional SQLAlchemy integration for persistence. Enable it with:

pip install "axion-framework[db]"

Then use the shared db singleton from axion.db. All of this lives under a single, re-usable connection object — no separate engine wiring required.

Connecting

The framework reads DATABASE_URL from settings (default: sqlite:///db.sqlite3) and connects lazily on first use.

# settings.py
DATABASE_URL = "sqlite:///db.sqlite3"

In main.py, wire the database into the request lifecycle:

from axion import WebServer
from axion.db import db

app = WebServer(debug=True)
app.use_database()

use_database(url=None, database=None) registers the DatabaseMiddleware (a transaction wrapper) and exposes the given Database — or the shared axion.db.db singleton by default.

Defining models

Models subclass axion.db.models.Model, the declarative base:

from axion.db.models import Model
from sqlalchemy import String
from sqlalchemy.orm import Mapped, mapped_column


class User(Model):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(100))

All models share one Model.metadata, so tables are created together. By default tables are created automatically on first session use (AUTO_CREATE_TABLES = True). Disable it to manage schemas yourself:

# settings.py
AUTO_CREATE_TABLES = False

Request-scoped sessions

db.session returns a thread-local session. Each worker thread gets its own session, and DatabaseMiddleware commits on success, rolls back on failure, and removes the session after every request.

Use sessions directly:

from axion.db import db
from axion.db.models import Model


def get_users():
    return db.session.query(User).all()


def create_user(name):
    user = User(name=name)
    db.session.add(user)
    db.session.flush()
    return user


def delete_user(user_id):
    user = db.session.get(User, user_id)
    if user:
        db.session.delete(user)
    return user

Injecting a session into handlers

Annotate a handler argument with Session and axion injects the request’s session automatically. Combined with the transaction middleware, writes are committed for you:

from sqlalchemy.orm import Session

from axion import WebServer
from axion.http import Response

app = WebServer(debug=True)
app.use_database()


@app.route("/api/users", methods=["POST"])
def create_user(name: str, email: str, session: Session, response: Response):
    user = User(name=name, email=email)
    session.add(user)
    session.flush()
    response.json({"id": user.id, "name": user.name}, status=201)


@app.route("/api/users/<int:user_id>")
def get_user(user_id: int, session: Session, response: Response):
    user = session.get(User, user_id)
    if user is None:
        response.send("User not found", 404)
        return
    response.json({"id": user.id, "name": user.name})

session.flush() sends the INSERT and populates auto-generated keys; the commit happens automatically when the request succeeds. If the handler raises, everything rolls back and a 500 is returned.

SQLite in-memory databases

sqlite:///:memory: uses a StaticPool with check_same_thread=False, so an in-memory database is shared across the worker threads and survives across requests. Ideal for tests.

db.configure("sqlite:///:memory:")

Database API

Member Description
db.url / db.echo Connection settings
db.engine The SQLAlchemy Engine (connects on first use)
db.session The request-scoped, thread-local session
db.connect() Create the engine and session factory
db.configure(url=None, *, echo=None) Reconfigure, disposing any prior engine
db.create_all() Create missing tables for all models
db.drop_all() Drop all model tables
db.remove() Remove the current thread’s session
db.dispose() Close the engine and release pooled connections

use_database() re-uses the shared singleton, but you can pass your own:

app.use_database(url="sqlite:///:memory:", database=my_db)

Configuring the transaction middleware manually

use_database() registers DatabaseMiddleware for you. To customize:

from axion.db import DatabaseMiddleware

app.use(DatabaseMiddleware(database=db))

Next steps