本文目录导读：
在Python中实现CQRS（Command Query Responsibility Segregation）模式，我推荐一种实用且清晰的实现方式，以下是完整的实现方案：
基础架构
核心抽象层
from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import TypeVar, Generic, Optional, List
from datetime import datetime
import uuid
# 类型定义
T = TypeVar('T')
R = TypeVar('R')
# --- 命令相关 ---
class Command(ABC):
"""命令基类"""
@property
@abstractmethod
def command_id(self) -> str:
pass
class CommandHandler(ABC, Generic[T]):
"""命令处理器基类"""
@abstractmethod
def handle(self, command: T) -> None:
pass
# --- 查询相关 ---
class Query(ABC):
"""查询基类"""
pass
class QueryResult(Generic[R]):
"""查询结果包装"""
def __init__(self, data: R, total: Optional[int] = None):
self.data = data
self.total = total
class QueryHandler(ABC, Generic[T, R]):
"""查询处理器基类"""
@abstractmethod
def handle(self, query: T) -> QueryResult[R]:
pass
# --- 事件相关 ---
@dataclass
class Event:
"""事件基类"""
event_id: str = None
timestamp: datetime = None
def __post_init__(self):
self.event_id = str(uuid.uuid4())
self.timestamp = datetime.utcnow()
class EventHandler(ABC, Generic[T]):
"""事件处理器基类"""
@abstractmethod
def handle(self, event: T) -> None:
pass
命令总线实现
class CommandBus:
"""命令总线 - 负责路由命令到对应处理器"""
def __init__(self):
self._handlers: dict = {}
def register(self, command_type: type, handler: CommandHandler):
"""注册命令处理器"""
self._handlers[command_type] = handler
def dispatch(self, command: Command) -> None:
"""分发命令到对应处理器"""
handler = self._handlers.get(type(command))
if not handler:
raise ValueError(f"No handler registered for {type(command).__name__}")
try:
handler.handle(command)
except Exception as e:
# 这里可以添加日志、重试逻辑等
raise CommandExecutionError(f"Command {command.command_id} failed: {str(e)}") from e
class CommandExecutionError(Exception):
"""命令执行异常"""
pass
查询总线实现
class QueryBus:
"""查询总线 - 负责路由查询到对应处理器"""
def __init__(self):
self._handlers: dict = {}
def register(self, query_type: type, handler: QueryHandler):
"""注册查询处理器"""
self._handlers[query_type] = handler
def ask(self, query: Query) -> QueryResult:
"""执行查询并返回结果"""
handler = self._handlers.get(type(query))
if not handler:
raise ValueError(f"No handler registered for {type(query).__name__}")
try:
return handler.handle(query)
except Exception as e:
raise QueryExecutionError(f"Query {type(query).__name__} failed: {str(e)}") from e
class QueryExecutionError(Exception):
"""查询执行异常"""
pass
事件总线实现
class EventBus:
"""事件总线 - 处理领域事件的发布和订阅"""
def __init__(self):
self._handlers: dict = {}
def register(self, event_type: type, handler: EventHandler):
"""注册事件处理器（支持多个处理器）"""
if event_type not in self._handlers:
self._handlers[event_type] = []
self._handlers[event_type].append(handler)
def publish(self, event: Event) -> None:
"""发布事件到所有注册的处理器"""
handlers = self._handlers.get(type(event), [])
for handler in handlers:
try:
handler.handle(event)
except Exception as e:
# 事件处理不应影响主流程
import logging
logging.error(f"Event handler {type(handler).__name__} failed: {str(e)}")
实战示例 - 用户管理
领域模型
@dataclass
class User:
user_id: str
username: str
email: str
created_at: datetime
updated_at: datetime
is_active: bool = True
# --- 命令定义 ---
@dataclass
class CreateUserCommand(Command):
username: str
email: str
password: str
@property
def command_id(self) -> str:
return str(uuid.uuid4())
@dataclass
class UpdateUserCommand(Command):
user_id: str
username: Optional[str] = None
email: Optional[str] = None
@property
def command_id(self) -> str:
return str(uuid.uuid4())
# --- 查询定义 ---
@dataclass
class GetUserQuery(Query):
user_id: str
@dataclass
class ListUsersQuery(Query):
page: int = 1
page_size: int = 20
search_term: Optional[str] = None
# --- 事件定义 ---
@dataclass
class UserCreatedEvent(Event):
user_id: str
username: str
email: str
@dataclass
class UserUpdatedEvent(Event):
user_id: str
changes: dict
存储层（简化版）
class UserRepository:
"""用户仓储 - 命令模型的存储"""
def __init__(self):
self._users: dict = {}
def save(self, user: User) -> None:
self._users[user.user_id] = user
def find_by_id(self, user_id: str) -> Optional[User]:
return self._users.get(user_id)
def delete(self, user_id: str) -> None:
self._users.pop(user_id, None)
class UserReadModel:
"""用户读模型 - 优化后的查询视图"""
def __init__(self):
self._users: dict = {}
def save(self, user: User) -> None:
# 可以在这里添加索引优化
self._users[user.user_id] = user
def find_by_id(self, user_id: str) -> Optional[dict]:
user = self._users.get(user_id)
if user:
return {
"user_id": user.user_id,
"username": user.username,
"email": user.email,
"created_at": user.created_at.isoformat()
}
return None
def list_all(self, page: int = 1, page_size: int = 20,
search_term: str = None) -> tuple:
"""返回 (数据列表, 总数)"""
users = list(self._users.values())
if search_term:
users = [u for u in users if search_term.lower() in u.username.lower()]
total = len(users)
start = (page - 1) * page_size
end = start + page_size
page_users = users[start:end]
return [
{
"user_id": u.user_id,
"username": u.username,
"email": u.email
} for u in page_users
], total
命令处理器
class CreateUserHandler(CommandHandler[CreateUserCommand]):
def __init__(self, user_repo: UserRepository, event_bus: EventBus):
self.user_repo = user_repo
self.event_bus = event_bus
def handle(self, command: CreateUserCommand) -> None:
# 验证业务规则
if len(command.password) < 8:
raise ValueError("Password must be at least 8 characters")
# 创建用户
user = User(
user_id=str(uuid.uuid4()),
username=command.username,
email=command.email,
created_at=datetime.utcnow(),
updated_at=datetime.utcnow()
)
# 持久化
self.user_repo.save(user)
# 发布事件
event = UserCreatedEvent(
user_id=user.user_id,
username=user.username,
email=user.email
)
self.event_bus.publish(event)
class UpdateUserHandler(CommandHandler[UpdateUserCommand]):
def __init__(self, user_repo: UserRepository, event_bus: EventBus):
self.user_repo = user_repo
self.event_bus = event_bus
def handle(self, command: UpdateUserCommand) -> None:
user = self.user_repo.find_by_id(command.user_id)
if not user:
raise ValueError(f"User {command.user_id} not found")
changes = {}
if command.username and command.username != user.username:
changes['username'] = command.username
user.username = command.username
if command.email and command.email != user.email:
changes['email'] = command.email
user.email = command.email
if changes:
user.updated_at = datetime.utcnow()
self.user_repo.save(user)
event = UserUpdatedEvent(
user_id=user.user_id,
changes=changes
)
self.event_bus.publish(event)
查询处理器
class GetUserHandler(QueryHandler[GetUserQuery, dict]):
def __init__(self, read_model: UserReadModel):
self.read_model = read_model
def handle(self, query: GetUserQuery) -> QueryResult[dict]:
user = self.read_model.find_by_id(query.user_id)
if not user:
return QueryResult(data=None)
return QueryResult(data=user)
class ListUsersHandler(QueryHandler[ListUsersQuery, list]):
def __init__(self, read_model: UserReadModel):
self.read_model = read_model
def handle(self, query: ListUsersQuery) -> QueryResult[list]:
data, total = self.read_model.list_all(
page=query.page,
page_size=query.page_size,
search_term=query.search_term
)
return QueryResult(data=data, total=total)
事件处理器（同步读模型）
class UserCreatedEventHandler(EventHandler[UserCreatedEvent]):
def __init__(self, read_model: UserReadModel):
self.read_model = read_model
def handle(self, event: UserCreatedEvent) -> None:
# 更新读模型
user = User(
user_id=event.user_id,
username=event.username,
email=event.email,
created_at=event.timestamp,
updated_at=event.timestamp
)
self.read_model.save(user)
class UserUpdatedEventHandler(EventHandler[UserUpdatedEvent]):
def __init__(self, read_model: UserReadModel):
self.read_model = read_model
def handle(self, event: UserUpdatedEvent) -> None:
# 这里可以实现更复杂的同步逻辑
# 示例中简化处理
pass
依赖注入与使用示例
class CQRSContainer:
"""简化的DI容器"""
def __init__(self):
self._services = {}
self._initialize()
def _initialize(self):
# 初始化存储
user_repo = UserRepository()
read_model = UserReadModel()
# 初始化事件总线
event_bus = EventBus()
# 注册事件处理器
event_bus.register(UserCreatedEvent, UserCreatedEventHandler(read_model))
event_bus.register(UserUpdatedEvent, UserUpdatedEventHandler(read_model))
# 初始化命令总线
command_bus = CommandBus()
command_bus.register(CreateUserCommand, CreateUserHandler(user_repo, event_bus))
command_bus.register(UpdateUserCommand, UpdateUserHandler(user_repo, event_bus))
# 初始化查询总线
query_bus = QueryBus()
query_bus.register(GetUserQuery, GetUserHandler(read_model))
query_bus.register(ListUsersQuery, ListUsersHandler(read_model))
self._services['command_bus'] = command_bus
self._services['query_bus'] = query_bus
@property
def command_bus(self) -> CommandBus:
return self._services['command_bus']
@property
def query_bus(self) -> QueryBus:
return self._services['query_bus']
# 使用示例
def main():
container = CQRSContainer()
# 创建用户
create_cmd = CreateUserCommand(
username="john_doe",
email="john@example.com",
password="secure_password123"
)
container.command_bus.dispatch(create_cmd)
# 查询所有用户
list_query = ListUsersQuery(page=1, page_size=10)
result = container.query_bus.ask(list_query)
print(f"Total users: {result.total}")
for user in result.data:
print(f" - {user['username']} ({user['email']})")
进阶特性（可选实现）
装饰器风格注册
from functools import wraps
def command_handler(command_type):
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
return func(*args, **kwargs)
wrapper._is_command_handler = True
wrapper._command_type = command_type
return wrapper
return decorator
# 自动发现并注册
class AutoRegisterCommandBus(CommandBus):
def discover_and_register(self, module):
import inspect
for name, obj in inspect.getmembers(module):
if hasattr(obj, '_is_command_handler'):
handler = CommandHandlerAdapter(obj)
self.register(obj._command_type, handler)
异步支持
import asyncio
from concurrent.futures import ThreadPoolExecutor
class AsyncCommandBus:
def __init__(self, executor: ThreadPoolExecutor = None):
self._handlers = {}
self._executor = executor or ThreadPoolExecutor()
async def dispatch(self, command: Command) -> None:
loop = asyncio.get_event_loop()
await loop.run_in_executor(self._executor, self._sync_dispatch, command)
def _sync_dispatch(self, command: Command):
handler = self._handlers.get(type(command))
if handler:
handler.handle(command)
关键设计要点
- 命令与查询分离：命令修改状态但不返回值，查询返回值但不修改状态
- 单一职责：每个处理器只处理一种命令或查询
- 事件驱动：通过事件同步读写模型
- 依赖注入：便于测试和扩展
- 类型安全：使用泛型确保类型匹配
这个实现方案保持了核心的CQRS原则,同时考虑了Python的特性和实际开发需求，可以根据项目规模选择是否引入更复杂的消息队列或分布式实现。
