Skip to main content

Documentation Index

Fetch the complete documentation index at: https://mintlify.com/Indroic/HexCore/llms.txt

Use this file to discover all available pages before exploring further.

Queries are the read side of CQRS. They represent a request for data and carry no side effects — they never modify state, raise events, or touch the write model. HexCore enforces this with Query[TResult], a generic, frozen Pydantic model that declares its return type at the class level. Queries are always dispatched synchronously through InMemoryQueryBus.ask() and return the typed result directly to the caller.

​
The Query base class

# hexcore/domain/cqrs/queries.py
import typing as t
from pydantic import BaseModel, ConfigDict

TResult = t.TypeVar("TResult")

class Query(BaseModel, t.Generic[TResult]):
    """
    Base class for all queries.
    TResult declares the return type of the query.
    """
    model_config = ConfigDict(
        frozen=True,
        from_attributes=True,
    )
Query is a frozen Pydantic model — instances are immutable and can be safely cached or passed across layers. The generic type parameter TResult communicates the expected return type to both the bus and the handler, enabling end-to-end type inference.

​
Defining a query

# app/queries.py
from hexcore.domain.cqrs.queries import Query
from app.dtos import OrderDTO

class GetOrderByIdQuery(Query[OrderDTO]):
    order_id: str

class ListOrdersQuery(Query[list[OrderDTO]]):
    customer_id: str
    page: int = 1
    page_size: int = 20
The TResult type argument (OrderDTO, list[OrderDTO]) tells InMemoryQueryBus what type to expect from the handler, and your IDE will infer it automatically from await bus.ask(query).

​
AbstractQueryHandler

Each query type is served by exactly one handler that inherits from AbstractQueryHandler[TQuery, TResult]. 
# hexcore/domain/cqrs/handlers.py (simplified)
import abc
import typing as t

TQuery = t.TypeVar("TQuery", bound="Query[t.Any]")
TResult = t.TypeVar("TResult")

class AbstractQueryHandler(abc.ABC, t.Generic[TQuery, TResult]):
    @abc.abstractmethod
    async def handle(self, query: TQuery) -> TResult:
        """Execute the query and return the result."""
        raise NotImplementedError
​
TQuery
TypeVar
The concrete Query subclass this handler processes.
​
TResult
TypeVar
The return type. Must match the TResult declared in the Query class.

​
Writing a query handler

Query handlers should use read-optimised data sources — dedicated read models, projections, or direct SQL views — instead of the write-side aggregate repositories. 
# app/query_handlers.py
from hexcore.domain.cqrs.handlers import AbstractQueryHandler
from app.queries import GetOrderByIdQuery, ListOrdersQuery
from app.dtos import OrderDTO
from app.read_models import IOrderReadModel

class GetOrderByIdHandler(AbstractQueryHandler[GetOrderByIdQuery, OrderDTO]):
    def __init__(self, read_model: IOrderReadModel) -> None:
        self._read = read_model

    async def handle(self, query: GetOrderByIdQuery) -> OrderDTO:
        return await self._read.get_by_id(query.order_id)

class ListOrdersHandler(AbstractQueryHandler[ListOrdersQuery, list[OrderDTO]]):
    def __init__(self, read_model: IOrderReadModel) -> None:
        self._read = read_model

    async def handle(self, query: ListOrdersQuery) -> list[OrderDTO]:
        return await self._read.list_by_customer(
            customer_id=query.customer_id,
            page=query.page,
            page_size=query.page_size,
        )

​
Registering and asking

1
Register query handlers in HandlerRegistry
2
from hexcore.application.cqrs.registry import HandlerRegistry
from app.queries import GetOrderByIdQuery, ListOrdersQuery
from app.query_handlers import GetOrderByIdHandler, ListOrdersHandler
from app.read_models import SqlOrderReadModel

read_model = SqlOrderReadModel(session)

registry = (
    HandlerRegistry()
    .register_query_handler(GetOrderByIdQuery, GetOrderByIdHandler(read_model))
    .register_query_handler(ListOrdersQuery, ListOrdersHandler(read_model))
)
3
Build InMemoryQueryBus
4
from hexcore.application.cqrs.in_memory_buses import InMemoryQueryBus

query_bus = InMemoryQueryBus(registry=registry)
5
A MiddlewarePipeline is optional for queries. If you pass one, middlewares such as LoggingMiddleware will wrap each ask() call.
6
from hexcore.application.cqrs.pipeline import MiddlewarePipeline
from hexcore.infrastructure.cqrs.middlewares import LoggingMiddleware

pipeline = MiddlewarePipeline().add(LoggingMiddleware())
query_bus = InMemoryQueryBus(registry=registry, pipeline=pipeline)
7
Ask the bus
8
from app.queries import GetOrderByIdQuery, ListOrdersQuery

# Single result
order: OrderDTO = await query_bus.ask(GetOrderByIdQuery(order_id="ord-123"))

# Paginated list
orders: list[OrderDTO] = await query_bus.ask(
    ListOrdersQuery(customer_id="cust-42", page=2, page_size=10)
)

​
Complete example

from __future__ import annotations
import asyncio
from dataclasses import dataclass

from hexcore.domain.cqrs.queries import Query
from hexcore.domain.cqrs.handlers import AbstractQueryHandler
from hexcore.application.cqrs.registry import HandlerRegistry
from hexcore.application.cqrs.in_memory_buses import InMemoryQueryBus

# ── DTO ──────────────────────────────────────────────────────────
@dataclass
class ProductDTO:
    product_id: str
    name: str
    price: float

# ── Query ────────────────────────────────────────────────────────
class GetProductQuery(Query[ProductDTO]):
    product_id: str

# ── In-memory stub store ─────────────────────────────────────────
_PRODUCTS = {
    "prod-1": ProductDTO("prod-1", "Apple", 0.99),
    "prod-2": ProductDTO("prod-2", "Banana", 0.49),
}

# ── Handler ──────────────────────────────────────────────────────
class GetProductHandler(AbstractQueryHandler[GetProductQuery, ProductDTO]):
    async def handle(self, query: GetProductQuery) -> ProductDTO:
        if query.product_id not in _PRODUCTS:
            raise KeyError(f"Product not found: {query.product_id}")
        return _PRODUCTS[query.product_id]

# ── Wiring ───────────────────────────────────────────────────────
registry = HandlerRegistry()
registry.register_query_handler(GetProductQuery, GetProductHandler())

bus = InMemoryQueryBus(registry=registry)

# ── Usage ────────────────────────────────────────────────────────
async def main():
    product = await bus.ask(GetProductQuery(product_id="prod-1"))
    print(f"{product.name}: ${product.price}")

asyncio.run(main())

​
FastAPI integration

# app/routers/products.py
from fastapi import APIRouter, Depends
from hexcore.domain.cqrs.buses import IQueryBus
from app.dependencies import get_query_bus
from app.queries import GetProductQuery
from app.dtos import ProductDTO

router = APIRouter(prefix="/products")

@router.get("/{product_id}", response_model=ProductDTO)
async def get_product(
    product_id: str,
    bus: IQueryBus = Depends(get_query_bus),
):
    return await bus.ask(GetProductQuery(product_id=product_id))
Queries are always synchronous and local in HexCore. There is no async background dispatch, no queue, and no eventual consistency — bus.ask() blocks until the handler returns a value. This is intentional: read operations must return data to the caller immediately. Never use a command bus for read operations, even if a read seems “expensive” — optimise the read model instead.

​
Queries vs. Commands at a glance

AspectCommandQuery
Base classCommandQuery[TResult]
Return typeAny / NoneAlways typed TResult
Side effectsYes (state changes)None
Background routingSupported (@background_command)❌ Never
Bus methodbus.dispatch(cmd)bus.ask(query)
MiddlewareFull pipelineOptional pipeline

Build docs developers (and LLMs) love

Get started for freeTalk to us