cnife 分层架构总览
API Router → Service → Domain → Repository → DB
↓
Client → 外部服务
编码须严格遵守每层的职责边界，禁止跨层调用或职责混淆。
分层编码规范
API 路由层
文件：api/v1/xxx.py
- 使用 FastAPI Router 或 Django View
- Pydantic 模型做请求/响应校验和序列化
- 异常统一由异常处理器转换，路由层不 catch
- 不写业务逻辑
# ✅ 正确：Router 只做路由和校验
from fastapi import APIRouter, Depends, HTTPException
from pydantic import BaseModel
router = APIRouter(prefix="/orders", tags=["orders"])
class OrderResponse(BaseModel):
id: int
status: str
amount: Decimal
@router.get("/{order_id}", response_model=OrderResponse)
def get_order(order_id: int, service: OrderService = Depends()):
"""查询订单"""
order = service.get_order(order_id)
if order is None:
raise HTTPException(status_code=404, detail="order not found")
return order
Service 层
文件：services/xxx_service.py
- 业务逻辑编排，事务控制
- 一个方法对应一个完整的业务用例
- 跨 Domain 对象的协调在此层完成
# ✅ 正确：Service 编排业务逻辑
from sqlalchemy.orm import Session
class OrderService:
def __init__(self, db: Session, payment_client: PaymentClient):
self.db = db
self.payment_client = payment_client
def create_order(self, user_id: int, items: list[OrderItem]) -> Order:
total = sum(item.price * item.qty for item in items)
order = Order(user_id=user_id, total_amount=total, status="PENDING")
self.db.add(order)
self.db.flush()
self.payment_client.charge(user_id=user_id, amount=total)
return order
Domain 层
文件：domain/xxx.py
- 核心业务逻辑和领域规则
- 业务校验（状态机转换、金额计算）
- 不感知外部服务、不处理 HTTP 细节
# ✅ 正确：Domain 专注业务规则
from dataclasses import dataclass
from decimal import Decimal
@dataclass
class Order:
id: int | None = None
user_id: int = 0
total_amount: Decimal = Decimal("0.00")
status: str = "PENDING"
def can_cancel(self) -> bool:
return self.status in {"PENDING", "PAID"}
def apply_discount(self, rate: Decimal) -> None:
if not Decimal("0") < rate < Decimal("1"):
raise ValueError("discount rate must be between 0 and 1")
self.total_amount = round(self.total_amount * rate, 2)
Repository 层
文件：repositories/xxx_repo.py
- 数据持久化操作
- Repository 封装 SQLAlchemy 细节，对外提供领域友好的接口
# ✅ 正确：Repository 封装持久化细节
from sqlalchemy.orm import Session
class OrderRepository:
def __init__(self, db: Session):
self.db = db
def find_by_id(self, order_id: int) -> OrderModel | None:
return self.db.query(OrderModel).filter(OrderModel.id == order_id).first()
def find_by_user(self, user_id: int, limit: int = 20) -> list[OrderModel]:
return (
self.db.query(OrderModel)
.filter(OrderModel.user_id == user_id)
.order_by(OrderModel.id.desc())
.limit(limit)
.all()
)
def save(self, order: OrderModel) -> None:
self.db.add(order)
Client 层（适配层）
文件：clients/xxx_client.py
- 外部 HTTP/gRPC 服务调用的封装
- 超时设置、重试策略、降级方案
# ✅ 正确：Client 封装外部调用
import httpx
from decimal import Decimal
class PaymentClient:
def __init__(self, base_url: str, timeout: float = 3.0):
self.client = httpx.Client(base_url=base_url, timeout=timeout)
def charge(self, user_id: int, amount: Decimal) -> dict:
try:
resp = self.client.post("/pay", json={"user_id": user_id, "amount": str(amount)})
resp.raise_for_status()
return resp.json()
except httpx.TimeoutException:
return {"status": "fallback", "message": "payment service timeout"}
except httpx.HTTPStatusError:
return {"status": "failed", "message": "payment rejected"}
通用约束
- 价格字段：
Decimal类型，禁止float - 外部调用：必须设置超时（≤5s）和降级方案
- 异步任务：耗时操作委托 Celery/Dramatiq worker
- 日志：关键链路使用
structlog或logging，异常场景打印完整 traceback - 类型注解：所有函数参数和返回值必须标注类型
产出物
- 代码变更（按 tasks.md 逐项完成）
- 编码报告：
coding/coding_report_v1.md- 变更文件列表
- 每个文件的变更摘要
- 未完成的 Task 及其原因