Skip to main content

CQRS (Command Query Responsibility Segregation)

info

CQRS separates read and write operations for better scalability and maintainability.

What is CQRS?​

CQRS (Command Query Responsibility Segregation) is a pattern that separates:

  • Commands: Write operations that change state
  • Queries: Read operations that retrieve data

Commands in Fluvius​

Commands represent intentions to change state. They are processed by aggregates and generate events.

Creating Commands​

# Create a command
command = domain.create_command(
    'create-user',
    payload={'name': 'John Doe', 'email': '[email protected]'},
    aggroot=('user', user_id)
)

Processing Commands​

# Process a single command
response = await domain.process_command(command)

# Process multiple commands
async for response in domain.command_processor.process(*commands):
    print(response)

Command Handlers​

Commands are handled by methods decorated with @command:

from fluvius.domain import command

class UserDomain(Domain):
    @command('create-user')
    async def handle_create_user(self, bundle):
        aggregate = self.get_aggregate()
        return await aggregate.create_user(
            name=bundle.payload.name,
            email=bundle.payload.email
        )

Queries in Fluvius​

Queries retrieve data without modifying state. They use the State Manager for read operations.

Simple Queries​

# Fetch a single entity
user = await domain.statemgr.fetch('user', user_id)

# Find entities
users = await domain.statemgr.find('user', active=True)

# Find one entity
user = await domain.statemgr.find_one('user', email='[email protected]')

Advanced Queries​

from fluvius.query import QueryManager

# Use Query Manager for complex queries
query_manager = QueryManager()
query = query_manager.build_query('user', {
    'filter': {'active': True},
    'sort': [{'field': 'name', 'order': 'asc'}],
    'limit': 10
})

users = await domain.statemgr.query(query)

Benefits of CQRS​

  1. Scalability: Read and write can scale independently
  2. Performance: Optimize read and write paths separately
  3. Flexibility: Different models for reads and writes
  4. Maintainability: Clear separation of concerns

Command-Query Separation Rules​

  1. Commands don't return data: They return success/failure status
  2. Queries don't modify state: They only read data
  3. Use events for side effects: Don't perform side effects in queries

Example: User Management​

# Command: Create user
command = domain.create_command('create-user', {
    'name': 'John Doe',
    'email': '[email protected]'
})
response = await domain.process_command(command)

# Query: Get user
user = await domain.statemgr.fetch('user', user_id)

# Query: List active users
active_users = await domain.statemgr.find('user', active=True)

Next Steps​