python_cqrs_core.query¶

Query interfaces for CQRS pattern.

Extracted from GridFlow backend/src/apps/token_generator/application/common/ports_interfaces/query.py

Attributes¶

`TQuery`
`TResult`

Classes¶

`IQuery`	Base query interface (read operations).
`IQueryHandler`	Query handler interface.

Module Contents¶

python_cqrs_core.query.TQuery¶

python_cqrs_core.query.TResult¶

class python_cqrs_core.query.IQuery¶

Bases: abc.ABC

Base query interface (read operations).

Queries read state without modifying it. Use this as a marker interface for all queries.

Example:

>>> from pydantic import BaseModel
>>>
>>> class GetUserQuery(BaseModel, IQuery):
...     user_id: int

class python_cqrs_core.query.IQueryHandler¶

Bases: abc.ABC, Generic[TQuery, TResult]

Query handler interface.

Handles a specific query type and returns a result.

Example:

>>> class GetUserHandler(IQueryHandler[GetUserQuery, User]):
...     async def handle(self, query: GetUserQuery) -> User:
...         # Fetch user logic
...         return user

abstract handle(query: TQuery) → TResult¶

Async:

Handle the query.

Args:: query: Query to handle
Returns:: Query result