coding-skill

name	coding-skill
description	分层编码实现规范，覆盖 API 路由→Service→Domain→Repository→Client 全链路
license	MIT
compatibility	opencode

分层架构总览

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 及其原因