技能版本 1.0.0
cqrs-implementation
Implement Command Query Responsibility Segregation for scalable architectures. Use when separating read and write models, optimizing query performance, or building event-sourced systems.
价格
免费
- 安装：免费
在组织开通时一次性收取；再次读取清单或下载不重复收费。
以上为公开标价。你的组织可能有协议价，以控制台显示的价格为准。
如何使用
为组织安装
开通即安装：组织固定使用开通时的版本（当前为 1.0.0）。升级到新版本需要取消后重新开通。
创建凭证
客户端使用组织凭证读取；绑定到数字人后，数字人使用绑定凭证按需加载。请求中携带：
Authorization: Bearer $ZGI_MARKET_KEY读取清单与安装包
清单列出版本、文件和校验值；安装包是一个 zip 文件。
读取清单 curl "https://market-api.zgi.im/v1/market/skills/cqrs-implementation/manifest" \ -H "Authorization: Bearer $ZGI_MARKET_KEY"下载安装包 curl "https://market-api.zgi.im/v1/market/skills/cqrs-implementation/package" \ -H "Authorization: Bearer $ZGI_MARKET_KEY" \ -o cqrs-implementation.zip读取清单和下载安装包不计入用量。下载后请用响应头中的 SHA-256 校验文件。
技能说明
- 名称
- cqrs-implementation
- 说明
- Implement Command Query Responsibility Segregation for scalable architectures. Use when separating read and write models, optimizing query performance, or building event-sourced systems.
仅指令
技能包内没有脚本文件：这个技能是供模型阅读的文本。
SKILL.md 节选
# CQRS Implementation
Comprehensive guide to implementing CQRS (Command Query Responsibility Segregation) patterns.
## When to Use This Skill
- Separating read and write concerns
- Scaling reads independently from writes
- Building event-sourced systems
- Optimizing complex query scenarios
- Different read/write data models needed
- High-performance reporting requirements
## Core Concepts
### 1. CQRS Architecture
```
┌─────────────┐
│ Client │
└──────┬──────┘
│
┌────────────┴────────────┐
│ │
▼ ▼
┌─────────────┐ ┌─────────────┐
│ Commands │ │ Queries │
│ API │ │ API │
└──────┬──────┘ └──────┬──────┘
│ │
▼ ▼
┌─────────────┐ ┌─────────────┐
│ Command │ │ Query │
│ Handlers │ │ Handlers │
└──────┬──────┘ └──────┬──────┘
│ │
▼ ▼
┌─────────────┐ ┌─────────────┐
│ Write │─────────►│ Read │
│ Model │ Events │ Model │
└─────────────┘ └─────────────┘
```
### 2. Key Components
| Component | Responsibility |
| ------------------- | ------------------------------- |
| **Command** | Intent to change state |
| **Command Handler** | Validates and executes commands |
| **Event** | Record of state change |
| **Query** | Request for data |
| **Query Handler** | Retrieves data from read model |
| **Projector** | Updates read model from events |
## Templates
### Template 1: Command Infrastructure
```python
from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import TypeVar, Generic, Dict, Any, Type
from datetime import datetime
import uuid
# Command base
@dataclass
class Command:
command_id: str = None
timestamp: datetime = None
def __post_init__(self):
self.command_id = self.command_id or str(uuid.uuid4())
self.timestamp = self.timestamp or datetime.utcnow()
# Concrete commands
@dataclass
class CreateOrder(Command):
customer_id: str
items: list
shipping_address: dict
@dataclass
class AddOrderItem(Command):
order_id: str
product_id: str
quantity: int
price: float
@dataclass
class CancelOrder(Command):
order_id: str
reason: str
# Command handler base
T = TypeVar('T', bound=Command)
class CommandHandler(ABC, Generic[T]):
@abstractmethod
async def handle(self, command: T) -> Any:
pass
# Command bus
class CommandBus:
def __init__(self):
self._handlers: Dict[Type[Command], CommandHandler] = {}
def register(self, command_type: Type[Command], handler: CommandHandler):
self._handlers[command_type] = handler
async def dispatch(self, command: Command) -> Any:
handler = self._handlers.get(type(command))
if not handler:
raise ValueError(f"No handler for {type(command).__name__}")
return await handler.handle(command)
# Command handler implementation
class CreateOrderHandler(CommandHandler[CreateOrder]):SKILL.md 不止这些内容，安装后可阅读全文。
技能包内容
共 1 个文件，合计 15.7 KiB
- SKILL.md15.7 KiB
来源与审核
- 审核状态
- 尚未审核，内容来自上游来源
- 许可证
- 未说明
审核状态是平台对上架信息的记录，不构成对上游服务质量或可用性的担保。