本文目录导读：
在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的特性和实际开发需求，可以根据项目规模选择是否引入更复杂的消息队列或分布式实现。
版权声明：NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
