When to apply CQRS in a Python service

← Back to the section

Imagine an order service. Writing an order is strict: we check the stock, lock the item, save transactionally. Reading is different: we assemble a summary from six tables and return it as JSON. These are two completely different tasks, but often they are served by the same class and the same database session.

CQRS (Command Query Responsibility Segregation) says: separate the write model from the read model. In its simple form — two types of handlers with different transaction settings. In its complex form — physically different stores synchronized through events. Between these extremes is a spectrum, and moving along it is worthwhile only when there's a real reason.

Three tiers of CQRS

The pattern is not binary. There are three points on the scale, and you should start with the simplest:

Movement is always bottom-up — first markers, then a separate table, then a separate store. Jumping straight to the third tier "for future growth" is a common mistake.

The first tier: markers with no extra infrastructure

The simplest kind of CQRS is just to denote the type of operation with a separate class. Command changes state, Query only reads.

In Python this is a @dataclass(frozen=True) — an immutable object that describes an intention:

from dataclasses import dataclass
from app.core.order.domain.model import OrderId

@dataclass(frozen=True)
class ConfirmOrder:       # this is a command — it changes state
    order_id: OrderId
    idempotency_key: str

@dataclass(frozen=True)
class GetOrderSummary:    # this is a query — it only reads
    order_id: OrderId

An important point: the query handler must get a database session without write rights. Then an accidental session.add() simply won't be saved:

class GetOrderSummaryHandler:
    def __init__(self, orders: OrderRepository) -> None:
        self._orders = orders

    async def handle(self, query: GetOrderSummary) -> OrderSummaryView:
        return await self._orders.by_id_readonly(query.order_id)

What this gives in practice:

• Mypy and Pyright see the difference between Command and Query at the type level — mixing them up is an error.
• A read-only session protects against accidental changes in a query handler.
• Metrics read meaningfully: app_command_total{name="ConfirmOrder"} and app_query_total{name="GetOrderSummary"} are different counters.
• Validation differs: for a command it's strict (Pydantic with frozen=True), for a query it's light (checking page/size ranges).

Meanwhile both read and write go to the same database. No additional infrastructure.

The second tier: a separate table for reading

When the query handler starts assembling data from five or six tables on every request, real pain appears: slow joins, high CPU load, complex queries that are hard to optimize.

Example: OrderSummaryView is assembled from order, order_item, product, customer, payment, shipment. When loading the order history — all six joins every time.

The solution is a denormalized order_summary table in the same database. Not a new store, just a separate table with a ready-made structure for the API:

@dataclass(frozen=True)
class OrderSummaryView:
    order_id: int
    customer_name: str    # denormalized from customer
    status: str
    item_count: int       # precomputed
    total_amount: Decimal
    created_at: datetime

The read repository is declared as a separate Protocol:

from typing import Protocol
from app.core.order.domain.model import OrderId, CustomerId

class OrderViewRepository(Protocol):
    async def by_id(self, order_id: OrderId) -> OrderSummaryView: ...
    async def by_customer(
        self,
        customer_id: CustomerId,
        limit: int,
        offset: int,
    ) -> list[OrderSummaryView]: ...

The GET /customers/{id}/orders request is now a single SELECT by index, with no joins. Synchronization between the write tables and the read table goes through the outbox pattern inside the same service.

Typical signals for moving to this tier:

• A typical query assembles 5 or more joins.
• Heavy GROUP BY over millions of rows for a dashboard or analytics.
• The UI wants customer_name, total_items, last_status_change — and in the normalized schema this is assembled anew on every request.

The third tier: separate stores — only under measured pain

A full separation of stores is justified only in specific situations with real numbers:

A read-to-write ratio of 10:1 and above. A typical online store: one order is created, but the card is shown ten times in different scenarios (history, search, analytics, notifications). At such a ratio the read load dictates the architecture.

A fundamentally different data structure. Full-text search over descriptions with filters across twenty parameters and relevance ranking is not a relational task. ElasticSearch or OpenSearch with an inverted index solves it an order of magnitude more efficiently than PostgreSQL.

The read load exceeds the write database's capacity. PostgreSQL handles 5 thousand write operations per second, but 50 thousand read operations of the same volume already require caching or a separate store.

An example infrastructure for a product catalog:

• PostgreSQL with the Product aggregate and an outbox table for events.
• A Kafka relay publishes ProductPublished, ProductPriceChanged.
• An ElasticSearch index products with a denormalized structure; the consumer updates the document on each event.
• FastAPI: POST /products → command handler in PostgreSQL; GET /products?q=... → query handler in ElasticSearch.

class SearchProductsHandler:
    def __init__(self, products: ProductSearchRepository) -> None:
        self._products = products

    async def handle(self, query: SearchProducts) -> ProductSearchPage:
        return await self._products.search(
            text=query.text,
            filters=query.filters,
            page=query.page,
            size=query.size,
        )

ProductSearchRepository is a Protocol backed by an ElasticSearch adapter. The FastAPI controller knows nothing about the concrete store.

This is expensive infrastructure: two stores, monitoring for both, data that can temporarily diverge (eventual consistency), rebuild scripts needed on failures. It pays off only when a read replica and a cache can no longer cope with the measured load.

Common mistakes

A full separation of stores from the very start. A new service with no measured load is no reason to immediately build two stores. Start with markers, evolve by metrics.

Markers are there, but the read handler can write to the database. If the query handler gets an ordinary session with commit rights, the markers are just class names with no real protection. The session for a query handler must be read-only.

Separating stores with no explicit reason. Eventual consistency, synchronization, recovery scenarios — all of this is real complexity. There must be a concrete measured pain, not abstract "scalability".

In short

• CQRS has three tiers: markers → a separate read table → separate stores. Start at the bottom.
• Markers (Command / Query as @dataclass(frozen=True)) — minimal cost, maximal benefit for types and metrics.
• A query handler works with a read-only session — an accidental write simply won't be saved.
• A separate read table in the same DB — the solution for heavy joins without additional infrastructure.
• Separate stores — only at read:write ≥ 10:1, search/analytics tasks, or a measured throughput overrun.
• Full CQRS "for future growth" for a new service is complexity with no benefit.

What to read next

• Command side in Python — the write handler with a Command Protocol and a Unit of Work.
• Query side in Python — the read handler with a Query Protocol and a ViewRepository.
• Read-model — where to store and how to update the denormalized projection.