实战项目：会员管理系统（商业级标准）

# app/services/order_service.py — 幂等 + 行锁，本项目最值钱的 20 行
from sqlalchemy import select
from app.core.exceptions import BusinessError

async def purchase(self, req: OrderCreate, idempotency_key: str | None) -> OrderOut:
    async with self.db.begin():                       # 1. 开启事务
        # 2. 幂等检查：同 key 直接返回原订单（重试安全）
        if idempotency_key:
            existing = await self.orders.get_by_idem_key(idempotency_key)
            if existing:
                return existing                        # ORDER_IDEMPOTENT_REPLAY

        # 3. 行锁：锁定会员卡行，杜绝并发续卡互相覆盖
        card = await self.db.execute(
            select(MemberCard).where(MemberCard.member_id == req.member_id)
            .with_for_update()                         # ★ SELECT ... FOR UPDATE
        )
        ...
        # 4. 事务提交时全部生效；中途任何一步抛错 → 全部回滚

# app/core/audit.py — 中间件级审计，业务代码不写一行审计代码
from app.dao.audit_dao import AuditDao
from app.core.context import current_admin, current_request_id

class AuditMiddleware(BaseHTTPMiddleware):
    WRITE_METHODS = {"POST", "PUT", "PATCH", "DELETE"}

    async def dispatch(self, request, call_next):
        response = await call_next(request)
        if request.method in self.WRITE_METHODS and response.status_code < 400:
            admin = current_admin.get(None)
            if admin:
                await AuditDao.log(
                    admin_id=admin.id,
                    action=f"{request.path_params}.{request.method}",
                    request_id=current_request_id.get(),       # 关联日志链路
                    ip=request.client.host,
                    changes=getattr(request.state, "audit_changes", None),
                )
        return response

# app/routes/auth.py
@router.post("/refresh")
async def refresh(refresh_token: str):
    payload = decode_token(refresh_token)
    if await is_blacklisted(refresh_token):          # 已注销的 refresh 不可再用
        raise BusinessError("AUTH_TOKEN_EXPIRED", http_status=401)
    return {"access_token": create_access_token(int(payload["sub"]), payload["role"])}

@router.post("/logout")
async def logout(refresh_token: str, admin = Depends(get_current_admin)):
    await add_to_blacklist(refresh_token, ttl=7 * 24 * 3600)  # 剩余有效期
    return {"code": 0, "message": "ok"}

# app/config.py — 缺配置直接启动失败，而不是跑到一半才炸
from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    env: Literal["dev", "staging", "prod"]
    jwt_secret: str
    jwt_expire_hours: int = 15 / 60 + 1     # prod 用分钟级
    database_url: str = "sqlite+aiosqlite:///./data/member.db"
    docs_enabled: bool = True               # prod 里覆盖为 False

    @model_validator(mode="after")
    def check_prod(self):
        if self.env == "prod":
            assert len(self.jwt_secret) >= 32, "prod 环境 jwt_secret 必须 ≥32 位"
        return self

settings = Settings()   # env=prod 且配置缺失 → 进程启动即报错退出

# tests/integration/test_concurrent_purchase.py — 并发测试，商业级必须有
async def test_concurrent_renew_no_overwrite(client, member_with_card):
    async def renew(key):
        return await client.post("/api/v1/orders",
            json={"member_id": 1, "plan_id": 1},
            headers={"Authorization": "...", "Idempotency-Key": key})

    r1, r2 = await asyncio.gather(renew("k1"), renew("k2"))
    card = await get_card(1)
    # 两次续费都成功，到期时间 = 原到期 + 2 × 时长（而不是互相覆盖）
    assert card.expire_at == expected_base + timedelta(days=plan_days * 2)