← Back to the section

In systems with CQRS there are two sides: commands change state, queries only read. It sounds simple, but in practice developers often mix them — and end up with unnecessary load, test failures, and tangled logic. Let's work out how to organize the reading side in Python.

Why you can't read through the same repository as the writing side

A typical mistake: there is an OrderRepository that can save and load the Order aggregate. A developer uses the same one for API queries — loads the aggregate, maps it into a DTO, and returns it.

The problem here is not that "the pattern is violated". The problem is practical:

  • The aggregate loads everything the business logic needs: related entities, collections, nested objects. For rendering in the UI this is excessive.
  • A query can accidentally call a domain method and change state — in a read-only session that raises an error, without one it is silently lost.
  • It's hard to optimize: the indexes for writing and for reading are often different.

The reading side of CQRS solves this through separation: a separate query object, a separate repository, a separate session, and a separate data model for the API.

A Query is an immutable object with parameters

A query to the system is expressed as a frozen dataclass — an object with parameters that cannot be changed after creation.

# core/cqrs.py
from typing import Protocol, TypeVar

R_co = TypeVar("R_co", covariant=True)

class Query(Protocol[R_co]):
    ...

# core/order/query/get_order_summary.py
from dataclasses import dataclass

@dataclass(frozen=True)
class GetOrderSummaryQuery:
    order_id: str

# core/order/query/search_orders.py
from dataclasses import dataclass
from core.order.port.view import OrderStatus

@dataclass(frozen=True)
class SearchOrdersQuery:
    customer_id: str
    status: OrderStatus | None
    page: int
    page_size: int

frozen=True means that once created the object cannot be changed — this matters, because a query describes an intention to read data, not to change it. Names are written in the form Get…Query, Search…Query, List…Query.

The query handler — it reads and changes nothing

The handler takes a Query, turns to the ViewRepository, and returns a read-DTO. No domain methods, no commit.

# core/order/handler/get_order_summary_handler.py
from core.order.query.get_order_summary import GetOrderSummaryQuery
from core.order.port.view import OrderSummary
from core.order.port.out.order_view_repository import OrderViewRepository
from core.error import OrderNotFoundError

class GetOrderSummaryHandler:
    def __init__(self, order_view_repo: OrderViewRepository) -> None:
        self._repo = order_view_repo

    async def handle(self, query: GetOrderSummaryQuery) -> OrderSummary:
        result = await self._repo.find_by_id(query.order_id)
        if result is None:
            raise OrderNotFoundError(query.order_id)
        return result

# core/order/handler/search_orders_handler.py
from core.order.query.search_orders import SearchOrdersQuery
from core.order.port.view import OrderSummary, Page

class SearchOrdersHandler:
    def __init__(self, order_view_repo: OrderViewRepository) -> None:
        self._repo = order_view_repo

    async def handle(self, query: SearchOrdersQuery) -> Page[OrderSummary]:
        return await self._repo.search(
            customer_id=query.customer_id,
            status=query.status,
            page=query.page,
            page_size=query.page_size,
        )

In FastAPI the handler is wired in as a dependency:

# api/order/router.py
from fastapi import APIRouter, Depends
from core.order.query.get_order_summary import GetOrderSummaryQuery
from core.order.port.view import OrderSummary

router = APIRouter(prefix="/orders")

@router.get("/{order_id}/summary", response_model=OrderSummary)
async def get_order_summary(
    order_id: str,
    handler: GetOrderSummaryHandler = Depends(get_order_summary_handler),
) -> OrderSummary:
    return await handler.handle(GetOrderSummaryQuery(order_id=order_id))

A read-only session — protection at the database level

The reading side gets a separate session that cannot be committed. This is done explicitly with SET TRANSACTION READ ONLY — PostgreSQL will reject any UPDATE or INSERT at the server level, without waiting for an application error.

# infra/db/session.py
from typing import AsyncGenerator
from sqlalchemy import text
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker

async def get_read_session(
    session_factory: async_sessionmaker[AsyncSession],
) -> AsyncGenerator[AsyncSession, None]:
    async with session_factory() as session:
        await session.execute(text("SET TRANSACTION READ ONLY"))
        yield session
        # commit is intentionally absent

This dependency (get_read_session) is passed into the ViewRepository via DI. Physically mixing a read and a write session becomes impossible — they are different objects with different functions.

ViewRepository — a separate read-only Protocol

OrderViewRepository is a Protocol without save, commit, or delete methods. Only read methods.

# core/order/port/out/order_view_repository.py
from typing import Protocol
from core.order.port.view import OrderSummary, OrderListItem, OrderStatus, Page

class OrderViewRepository(Protocol):
    async def find_by_id(self, order_id: str) -> OrderSummary | None: ...
    async def search(
        self,
        customer_id: str,
        status: OrderStatus | None,
        page: int,
        page_size: int,
    ) -> Page[OrderSummary]: ...
    async def find_recent_by_customer(
        self, customer_id: str, limit: int
    ) -> list[OrderListItem]: ...

The implementation in infra/persistence/ builds SQL queries directly:

# infra/persistence/order_view_repository_impl.py
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from core.order.port.view import OrderSummary, OrderStatus

class SqlAlchemyOrderViewRepository:
    def __init__(self, session: AsyncSession) -> None:
        self._session = session

    async def find_by_id(self, order_id: str) -> OrderSummary | None:
        from infra.persistence.models import OrderModel, CustomerModel
        stmt = (
            select(
                OrderModel.id,
                OrderModel.status,
                CustomerModel.name.label("customer_name"),
                OrderModel.total_amount,
                OrderModel.item_count,
                OrderModel.created_at,
                OrderModel.updated_at,
            )
            .join(CustomerModel, CustomerModel.id == OrderModel.customer_id)
            .where(OrderModel.id == order_id)
        )
        row = (await self._session.execute(stmt)).one_or_none()
        if row is None:
            return None
        return OrderSummary(
            order_id=str(row.id),
            status=OrderStatus(row.status),
            customer_name=row.customer_name,
            total_amount=row.total_amount,
            item_count=row.item_count,
            created_at=row.created_at,
            updated_at=row.updated_at,
        )

If there is a denormalized order_summary table (populated from events), the query becomes even simpler — without the JOIN.

Read-DTO — a data model shaped for the API, not for the aggregate

A read-DTO is a Pydantic model whose structure is dictated by what the UI or API needs, not by how the aggregate is arranged.

# core/order/port/view.py
from __future__ import annotations
from datetime import datetime
from decimal import Decimal
from enum import StrEnum
from pydantic import BaseModel

class OrderStatus(StrEnum):
    PENDING = "PENDING"
    CONFIRMED = "CONFIRMED"
    SHIPPED = "SHIPPED"
    DELIVERED = "DELIVERED"
    CANCELLED = "CANCELLED"

class OrderSummary(BaseModel):
    model_config = {"frozen": True}

    order_id: str
    status: OrderStatus
    customer_name: str        # denormalized — no separate JOIN to customer
    total_amount: Decimal
    item_count: int           # number of line items, not a list of objects
    created_at: datetime
    updated_at: datetime

class OrderListItem(BaseModel):
    model_config = {"frozen": True}

    order_id: str
    status: OrderStatus
    total_amount: Decimal
    created_at: datetime

A few important decisions here:

  • customer_name is denormalized — the customer's name is stored right in the order row, no separate query to the customers table is needed.
  • item_count instead of list[OrderItem] — to display "5 items" in a list a number is enough; loading all the order line items is more expensive.
  • model_config = {"frozen": True} — Pydantic v2, the instance cannot be changed after creation.
  • StrEnum — the enum values serialize as strings, which is convenient for a JSON API.

Common mistake: calling domain methods in a query handler

# Wrong — the query calls a domain method
async def handle(self, query: GetOrderSummaryQuery) -> OrderSummary:
    order = await self._order_repo.by_id(query.order_id)  # loading the aggregate
    if order.should_be_archived():          # domain method in a read
        await order.archive()               # mutation in a read-only session
    return OrderSummary(order_id=str(order.id), ...)

What's wrong here:

  • SET TRANSACTION READ ONLY will raise an error from PostgreSQL on an attempt at INSERT/UPDATE. Without a read-only session the mutation is silently lost — there is no commit.
  • The reading side must not change state. If stale orders need to be archived, that is a separate command (ArchiveStaleOrdersCommand) with its own handler.

Correct: the query handler turns only to the ViewRepository and returns a read-DTO. No aggregate, no domain methods.

What the file structure looks like

core/
└── order/
    ├── domain/
    │   ├── order.py               # aggregate
    │   └── order_item.py
    ├── port/
    │   ├── out/
    │   │   ├── order_repository.py       # write-side (aggregate)
    │   │   └── order_view_repository.py  # read-side
    │   └── view.py                       # read-DTO
    └── handler/
        ├── confirm_order_handler.py      # command handler
        └── get_order_summary_handler.py  # query handler

In short

  • A Query is a @dataclass(frozen=True) with read parameters. It does not change state.
  • The query handler turns only to the ViewRepository and returns a read-DTO. No aggregate.
  • A read-only session (SET TRANSACTION READ ONLY) — PostgreSQL blocks UPDATE/INSERT at the server level.
  • The ViewRepository is a separate Protocol without save/commit. Only read methods.
  • A read-DTO (a Pydantic BaseModel with frozen=True) — a structure shaped for the API: denormalized fields, numbers instead of collections.
  • You must not load the aggregate in a query handler and map it into a DTO — that nullifies the benefit of the separation.
  • You must not call domain methods in a query handler — it is read-only, there must be no side effects.
  • Command side — the writing half: a handler through the aggregate and Unit of Work.
  • Read-model — where and in what form to store read data.
  • Sync via events — how the read table is populated from write-side events.
  • When CQRS is justified — when a simple split is enough and when a full split is needed.