coding-skill

Layered Architecture Overview

API Router → Service → Domain → Repository → DB
    ↓
  Client → 外部服务

Coding must strictly comply with the responsibility boundaries of each layer; cross-layer calls or confusion of responsibilities are prohibited.

---

Layered Coding Specification

API Route Layer

Files：api/v1/xxx.py

• Use FastAPI Router or Django View
• Use Pydantic models for request/response validation and serialization
• Exceptions are uniformly converted by the exception handler; the route layer does not catch
• Do not write business logic

# ✅ 正确：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 Layer

Files：services/xxx_service.py

• Business logic orchestration, transaction control
• One method corresponds to one complete business use case
• Coordination across Domain objects is completed at this layer

# ✅ 正确：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 Layer

Files：domain/xxx.py

• Core business logic and domain rules
• Business validation (state machine transitions, amount calculation)
• Does not perceive external services, does not handle HTTP details

# ✅ 正确：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 Layer

Files：repositories/xxx_repo.py

• Data persistence operations
• The Repository encapsulates SQLAlchemy details and provides a domain-friendly interface externally

# ✅ 正确：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 Layer (Adapter Layer)

Files：clients/xxx_client.py

• Encapsulation of external HTTP/gRPC service calls
• Timeout settings, retry strategies, fallback plans

# ✅ 正确：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"}

---

General Constraints

• Price field：Decimal type, prohibited float
• External calls: must set timeout (≤5s) and a fallback plan
• Asynchronous tasks: time-consuming operations are delegated to a Celery/Dramatiq worker
• Logging: critical paths use structlog or logging, and exception scenarios print the complete traceback
• Type annotations: all function parameters and return values must be annotated with types

---

Deliverables

• Code changes (completed item by item according to tasks.md)
• Coding report:coding/coding_report_v1.md
  • List of changed files
  • Change summary for each file
  • Uncompleted Tasks and their reasons