Commands & Queries (CQRS)

CQRS separates writes from reads at the application layer: commands change state, queries read it. axion does not ship CQRS primitives — it’s a web framework, not an application framework — but its routing, dependency injection, and database session make the pattern straightforward to apply. This page is a playbook, not a requirement.

Why apply it

For any app with more than a couple of routes, separating reads from writes keeps handler functions thin and makes behavior testable:

The shape of a module

my_app/
├── main.py
├── routers/
│   └── users.py          # HTTP: routes -> commands/queries
├── commands/
│   └── users.py          # create_user, update_user, delete_user
├── queries/
│   └── users.py          # get_user, list_users
├── models.py             # SQLAlchemy models (axion.db.Model)
└── config/
    └── settings.py

Commands

A command is a function that changes state. Give it the pieces it needs explicitly (session, input) and return a plain result:

# commands/users.py
from dataclasses import dataclass
from sqlalchemy.orm import Session

from models import User


@dataclass
class CreateUserCommand:
    name: str
    email: str


def create_user(session: Session, command: CreateUserCommand) -> User:
    user = User(name=command.name, email=command.email)
    session.add(user)
    session.flush()
    return user

The web layer parses the request, validates with a serializer, and dispatches:

# routers/users.py
from sqlalchemy.orm import Session

from axion import Router
from axion.http import Request, Response
from axion.serializer import Serializer

from commands.users import CreateUserCommand, create_user

users_router = Router()


class CreateUserSerializer(Serializer):
    name: str
    email: str


@users_router.route("/users", methods=["POST"])
def post_user(request: Request, session: Session, response: Response):
    serializer = CreateUserSerializer.from_data(request.json)
    if not serializer.is_valid():
        response.json(serializer.errors, 400)
        return
    user = create_user(session, CreateUserCommand(**serializer.validated_data))
    response.json({"id": user.id}, 201)

Because DatabaseMiddleware wraps every request in a transaction, the commit happens automatically when the route returns without error — commands don’t need to manage transactions.

Queries

A query is a read-only function. It takes a session and returns data:

# queries/users.py
from sqlalchemy.orm import Session

from models import User


def get_user(session: Session, user_id: int) -> User | None:
    return session.get(User, user_id)


def list_users(session: Session) -> list[User]:
    return list(session.query(User).all())

@users_router.route("/users/<int:user_id>")
def get_user(user_id: int, session: Session, response: Response):
    user = get_user(session, user_id)
    if user is None:
        response.send("Not found", 404)
        return
    response.json(UserSerializer.from_row(user).data)

When not to use it

CQRS (and its more advanced sibling, event sourcing) adds structure that pays off only when reads and writes genuinely diverge — different shapes, different frequencies, separate scaling. For a small CRUD app, a plain services layer (as in the example app) is simpler and fine. Apply the split when it earns its keep, not as a default.

Next steps