From python-architecture
CQRS pattern: separate write operations (commands) from read operations (queries)
How this skill is triggered — by the user, by Claude, or both
Slash command
/python-architecture:cqrsThe summary Claude sees in its skill listing — used to decide when to auto-load this skill
Command Query Responsibility Segregation: separate write operations (commands) from read operations (queries). Commands go through the service layer with full domain validation. Queries can bypass the domain model for performance.
Command Query Responsibility Segregation: separate write operations (commands) from read operations (queries). Commands go through the service layer with full domain validation. Queries can bypass the domain model for performance.
Route all state modifications through the service layer with full domain validation. Provide dedicated query functions in the views layer.
# Good
# service/company.py -- COMMAND (write)
async def activate_company(
company_id: CompanyId, repository: CompanyRepository
) -> None:
company = await repository.get(company_id)
company.activate() # Domain validation
await repository.save(company)
# views/company.py -- QUERY (read)
async def get_active_companies(session: AsyncSession) -> list[dict]:
result = await session.execute(
"SELECT id, name FROM companies WHERE active = true"
)
return [{"id": row.id, "name": row.name} for row in result]
# Bad -- mixing command and query
async def get_and_activate_company(company_id, repository):
company = await repository.get(company_id)
company.activate() # Side effect in query
await repository.save(company)
return company
Optimize read performance by accessing data stores without domain object instantiation.
# Good -- views/dashboard.py
@dataclass
class CompanySummary:
company_id: UUID
name: str
user_count: int
async def get_company_dashboard(session: AsyncSession) -> list[CompanySummary]:
result = await session.execute("""
SELECT c.id, c.name, COUNT(u.id) as user_count
FROM companies c LEFT JOIN users u ON u.company_id = c.id
GROUP BY c.id, c.name
""")
return [CompanySummary(**row) for row in result]
# Bad -- loading full domain objects for simple query
async def get_company_names(repository: CompanyRepository) -> list[str]:
companies = await repository.list()
return [company.name for company in companies]
Views are purely for data retrieval, returning simple DTOs or domain objects.
# Good -- pure query
async def get_company_statistics(company_id: CompanyId, session: AsyncSession) -> dict:
result = await session.execute(
"SELECT COUNT(*) as user_count FROM users WHERE company_id = :id",
{"id": company_id.value}
)
return {"user_count": result.scalar()}
# Bad -- side effects in view
async def get_company_and_log_access(company_id, session):
await session.execute(
"INSERT INTO access_log (company_id, accessed_at) VALUES (:id, NOW())", ...
)
return await session.get(Company, company_id.value)
npx claudepluginhub remihuguet/rems-buddy --plugin python-architectureGuides implementing CQRS to separate read and write models, optimize query performance, and build event-sourced systems.
Designs and implements CQRS patterns for scalable systems, covering logical separation, separate read models, event-sourced CQRS, and query optimization.
Separate command (write) and query (read) models for complex domains. Use when read/write patterns diverge significantly or when audit/consistency requirements demand immutability.