Command Query Responsibility Segregation (CQRS) lets your read and write operations scale independently, even if it sounds like you’re just adding complexity at first.
Let’s see it in action with a simple Python example for managing user profiles. We’ll use a basic web framework like Flask for the API endpoints.
# --- Commands ---
class UpdateUserProfileCommand:
def __init__(self, user_id: str, name: str = None, email: str = None):
self.user_id = user_id
self.name = name
self.email = email
class UpdateUserProfileCommandHandler:
def __init__(self, user_repository):
self.user_repository = user_repository
def handle(self, command: UpdateUserProfileCommand):
user = self.user_repository.get_by_id(command.user_id)
if user:
if command.name:
user.name = command.name
if command.email:
user.email = command.email
self.user_repository.save(user)
else:
raise ValueError(f"User with ID {command.user_id} not found.")
# --- Queries ---
class GetUserProfileQuery:
def __init__(self, user_id: str):
self.user_id = user_id
class GetUserProfileQueryHandler:
def __init__(self, user_read_repository):
self.user_read_repository = user_read_repository
def handle(self, query: GetUserProfileQuery):
return self.user_read_repository.get_by_id(query.user_id)
# --- Domain (Entities) ---
class User:
def __init__(self, id: str, name: str, email: str):
self.id = id
self.name = name
self.email = email
# --- Infrastructure (Repositories) ---
class InMemoryUserRepository:
def __init__(self):
self._users = {}
def get_by_id(self, user_id: str) -> User | None:
return self._users.get(user_id)
def save(self, user: User):
self._users[user.id] = user
class InMemoryUserReadRepository:
def __init__(self, user_repository: InMemoryUserRepository):
# In a real system, this would be a separate read-optimized store
self._user_repository = user_repository
def get_by_id(self, user_id: str) -> User | None:
return self._user_repository.get_by_id(user_id)
# --- Application (API/Entry Points) ---
from flask import Flask, request, jsonify
app = Flask(__name__)
# Setup repositories and handlers (dependency injection)
user_repository = InMemoryUserRepository()
user_read_repository = InMemoryUserReadRepository(user_repository) # Simplified for example
update_user_handler = UpdateUserProfileCommandHandler(user_repository)
get_user_handler = GetUserProfileQueryHandler(user_read_repository)
@app.route('/users/<user_id>', methods=['PUT'])
def update_user(user_id):
data = request.get_json()
command = UpdateUserProfileCommand(user_id, name=data.get('name'), email=data.get('email'))
try:
update_user_handler.handle(command)
return jsonify({"message": "User updated successfully"}), 200
except ValueError as e:
return jsonify({"error": str(e)}), 404
@app.route('/users/<user_id>', methods=['GET'])
def get_user(user_id):
query = GetUserProfileQuery(user_id)
user = get_user_handler.handle(query)
if user:
return jsonify({"id": user.id, "name": user.name, "email": user.email}), 200
else:
return jsonify({"error": "User not found"}), 404
# --- Example Usage ---
if __name__ == '__main__':
# Seed some data
user_repo_for_seed = InMemoryUserRepository()
user_read_repo_for_seed = InMemoryUserReadRepository(user_repo_for_seed)
update_user_handler_seed = UpdateUserProfileCommandHandler(user_repo_for_seed)
get_user_handler_seed = GetUserProfileQueryHandler(user_read_repo_for_seed)
initial_user = User("user123", "Alice Smith", "alice@example.com")
user_repo_for_seed.save(initial_user)
# Update global repositories for the app to use
user_repository = user_repo_for_seed
user_read_repository = user_read_repo_for_seed
update_user_handler = UpdateUserProfileCommandHandler(user_repository)
get_user_handler = GetUserProfileQueryHandler(user_read_repository)
# Simulate API calls (you'd run this with `flask run`)
print("--- Initial State ---")
print(get_user_handler.handle(GetUserProfileQuery("user123")))
print("\n--- Updating User ---")
update_command = UpdateUserProfileCommand("user123", name="Alice Wonderland")
update_user_handler.handle(update_command)
print(get_user_handler.handle(GetUserProfileQuery("user123")))
print("\n--- Getting User ---")
get_query = GetUserProfileQuery("user123")
print(get_user_handler.handle(get_query))
# To run the Flask app:
# app.run(debug=True)
This separates what you want to change (commands) from what you want to retrieve (queries). Commands are requests to mutate state, and queries are requests to read state. Each has its own handler. The "Clean Architecture" part comes in by keeping your core business logic (entities, use cases) free from infrastructure details like databases or web frameworks. Your User entity, for instance, doesn’t know or care if it’s being updated via a REST API or a gRPC service, or if its data is stored in PostgreSQL or Redis.
The real power emerges when your read and write models diverge. Imagine a user profile. The write model might need a full User object with all fields for validation and updates. The read model, however, might only need a UserProfileSummary object containing just the user’s name and avatar URL for a dashboard list. In our example, InMemoryUserRepository and InMemoryUserReadRepository are identical, but in a production system, they’d point to different data stores or query patterns. The write repository would talk to a transactional database optimized for writes (e.g., PostgreSQL), while the read repository could query a denormalized, read-optimized data store like Elasticsearch or a caching layer for blazing-fast retrieval.
Each command and query is a distinct object, making your intent explicit. A UpdateUserProfileCommand carries the intent to update a user and the data needed for that update. A GetUserProfileQuery carries the intent to retrieve a user and the identifier. This object-oriented approach makes the system more robust and easier to reason about than passing raw dictionaries or parameters around. The handlers act as the glue, taking these command/query objects and orchestrating the necessary domain logic and data access.
The "Clean Architecture" aspect is enforced by the dependency rule: outer layers depend on inner layers. Your API endpoints (outermost) depend on handlers, which depend on repositories and domain entities (innermost). The domain entities know nothing about handlers or repositories. This ensures your core business logic is testable in isolation, without needing a database or web server.
The most impactful, and often overlooked, benefit of CQRS is how it enables different scaling strategies for reads versus writes. If your application becomes read-heavy (e.g., a popular product catalog), you can scale out your read repositories independently. You might deploy multiple instances of your read data store, or implement read replicas, or add caching layers specifically for query operations. Conversely, if you have a write-intensive operation (e.g., processing high volumes of financial transactions), you can optimize and scale your write infrastructure without impacting read performance, and vice-versa. This granular control over scaling is a significant advantage over traditional monolithic architectures where read and write operations share the same resources.
The next logical step is to explore how event sourcing can be integrated with CQRS, allowing you to not only issue commands and queries but also to capture every state change as an immutable event.