通用后台框架开发心法 —— 从 likeadmin 学架构

收银台区：  珍珠奶茶收银员、柠檬茶收银员、奶盖茶收银员
制茶区：    珍珠奶茶制茶师、柠檬茶制茶师、奶盖茶制茶师
原料区：    珍珠奶茶原料员、柠檬茶原料员、奶盖茶原料员

珍珠奶茶组： 收银、制茶、管料三人背靠背
柠檬茶组：   收银、制茶、管料三人背靠背

server/
├── controller/           # 所有 Controller 堆一起
│   ├── UserController.php
│   ├── OrderController.php
│   └── GoodsController.php
├── logic/                # 所有 Logic 堆一起
│   ├── UserLogic.php
│   ├── OrderLogic.php
│   └── GoodsLogic.php
├── model/                # 所有 Model 堆一起
│   └── ...
└── validate/             # 所有 Validate 堆一起
    └── ...

server/
├── modules/                    # 业务模块——每个模块高内聚
│   ├── user/                   # 用户模块
│   │   ├── controller.py       # 接收请求
│   │   ├── service.py          # 业务逻辑
│   │   ├── repository.py       # 数据访问
│   │   ├── model.py            # 数据模型
│   │   ├── schema.py           # 请求/响应结构
│   │   └── router.py           # 路由注册
│   ├── order/                  # 订单模块
│   │   └── ...（同上结构）
│   └── goods/                  # 商品模块
│       └── ...
├── common/                     # 跨模块共享——低耦合的桥梁
│   ├── response.py             # 统一响应
│   ├── exceptions.py           # 统一异常
│   ├── middleware/              # 公共中间件
│   └── utils/                  # 按职责分组的工具类
│       ├── string_utils.py
│       ├── crypto_utils.py
│       └── date_utils.py
├── core/                       # 框架核心——与业务无关的基础设施
│   ├── config.py               # 配置管理
│   ├── database.py             # 数据库连接
│   ├── security.py             # 安全相关
│   └── dependencies.py         # 公共依赖
└── main.py                     # 应用入口

收银员（Controller）： 接单 + 帮忙校验订单 + 偶尔帮制茶师调配方
制茶师（Logic）：      大多数时候只喊"仓库拿红茶"，偶尔调配方
仓库+配方+标准（Model）：又管仓库、又定配方、又管标准——一人干三份活

┌──────────────────────────────────────────────────┐
│                  Controller 层                     │
│         接收请求、调度服务、返回响应                  │
│         （不写任何业务逻辑和数据访问）                 │
├──────────────────────────────────────────────────┤
│                   Schema 层                        │
│         定义输入输出的形状、验证规则                   │
│         （脏数据在这里被拦截，进不了业务层）            │
├──────────────────────────────────────────────────┤
│                   Service 层                       │
│         业务规则、流程编排、事务管理                   │
│         （整个框架的核心大脑）                        │
├──────────────────────────────────────────────────┤
│                 Repository 层                      │
│         数据存取、查询封装、分页                       │
│         （只跟数据库打交道，不碰业务逻辑）              │
├──────────────────────────────────────────────────┤
│                   Model 层                         │
│         表结构定义、字段约束                          │
│         （纯粹的数据结构，零逻辑）                     │
└──────────────────────────────────────────────────┘

# modules/user/controller.py
from fastapi import APIRouter, Depends, Query
from .schema import CreateUserRequest, UpdateUserRequest, UserResponse
from .service import UserService
from common.response import ApiResponse, PageData
from core.dependencies import PermissionChecker

router = APIRouter(prefix="/users", tags=["用户管理"])

@router.get("/", response_model=ApiResponse[PageData[UserResponse]],
            dependencies=[Depends(PermissionChecker("user:list"))])
async def list_users(
    page: int = Query(1, ge=1),
    size: int = Query(10, ge=1, le=100),
    keyword: str = Query(None),
    service: UserService = Depends()
):
    items, total = await service.list_users(page, size, keyword)
    return ApiResponse.success(PageData(items=items, total=total, page=page, size=size))

@router.post("/", response_model=ApiResponse[UserResponse],
             dependencies=[Depends(PermissionChecker("user:create"))])
async def create_user(data: CreateUserRequest, service: UserService = Depends()):
    user = await service.create_user(data)
    return ApiResponse.success(user, "创建成功")

@router.get("/{user_id}", response_model=ApiResponse[UserResponse])
async def get_user(user_id: int, service: UserService = Depends()):
    return ApiResponse.success(await service.get_user(user_id))

@router.put("/{user_id}", response_model=ApiResponse[UserResponse],
            dependencies=[Depends(PermissionChecker("user:update"))])
async def update_user(user_id: int, data: UpdateUserRequest, service: UserService = Depends()):
    return ApiResponse.success(await service.update_user(user_id, data), "更新成功")

@router.delete("/{user_id}", response_model=ApiResponse,
               dependencies=[Depends(PermissionChecker("user:delete"))])
async def delete_user(user_id: int, service: UserService = Depends()):
    await service.delete_user(user_id)
    return ApiResponse.success(msg="删除成功")

// modules/user/controller.js —— Fastify 版本
module.exports = async function (fastify, opts) {
  fastify.get('/users', {
    preHandler: [fastify.auth, fastify.requirePermission('user:list')],
    schema: { response: { 200: pagedUserResponseSchema } }
  }, async (request, reply) => {
    const { page = 1, size = 10, keyword } = request.query
    const { items, total } = await fastify.userService.listUsers(page, size, keyword)
    return ApiResponse.success(ApiResponse.pageData(items, total, page, size))
  })

  fastify.post('/users', {
    preHandler: [fastify.auth, fastify.requirePermission('user:create')],
    schema: { body: createUserSchema, response: { 200: userResponseSchema } }
  }, async (request, reply) => {
    const user = await fastify.userService.createUser(request.body)
    return ApiResponse.success(user, '创建成功')
  })
}

# modules/user/schema.py
from pydantic import BaseModel, Field, EmailStr, field_validator
from datetime import datetime
import re

class CreateUserRequest(BaseModel):
    """创建用户请求 —— 声明即验证"""
    username: str = Field(..., min_length=2, max_length=50, description="用户名")
    phone: str = Field(..., pattern=r'^1[3-9]\d{9}$', description="手机号")
    password: str = Field(..., min_length=6, max_length=20, description="密码")
    email: EmailStr | None = Field(None, description="邮箱")

    @field_validator('password')
    @classmethod
    def password_strength(cls, v: str) -> str:
        if not re.search(r'[A-Za-z]', v) or not re.search(r'[0-9]', v):
            raise ValueError('密码必须包含字母和数字')
        return v

class UpdateUserRequest(BaseModel):
    """更新用户请求 —— 所有字段可选"""
    username: str | None = Field(None, min_length=2, max_length=50)
    email: EmailStr | None = None
    avatar: str | None = None

class UserResponse(BaseModel):
    """用户信息响应 —— 返回给前端的数据结构"""
    id: int
    username: str
    phone: str
    email: str | None
    avatar: str | None
    is_active: bool
    created_at: datetime

    model_config = {"from_attributes": True}  # 自动从 ORM 对象转换

class UserListResponse(BaseModel):
    """用户列表响应"""
    items: list[UserResponse]
    total: int
    page: int
    size: int

# modules/user/service.py
from .repository import UserRepository
from common.exceptions import AppException, ErrorCode
from common.utils.crypto_utils import CryptoUtils

class UserService:
    """用户服务 —— 所有用户相关的业务规则在这里"""

    def __init__(self, repo: UserRepository = Depends()):
        self.repo = repo

    async def create_user(self, data) -> "User":
        """
        创建用户的完整业务流程：
        1. 检查手机号唯一性（业务规则）
        2. 密码加密（业务规则）
        3. 保存到数据库（调用仓库层）
        4. 发送欢迎短信（业务编排）
        """
        # 业务规则1：手机号唯一
        if await self.repo.find_by_phone(data.phone):
            raise AppException(ErrorCode.USER_PHONE_EXISTS, "该手机号已注册")

        # 业务规则2：用户名唯一
        if await self.repo.find_by_username(data.username):
            raise AppException(ErrorCode.USERNAME_EXISTS, "用户名已被占用")

        # 业务规则3：密码加密存储
        hashed_pwd = CryptoUtils.hash_password(data.password)

        # 调用仓库层保存（Service 不写 SQL）
        user = await self.repo.create(
            username=data.username,
            phone=data.phone,
            password=hashed_pwd,
            email=data.email,
        )

        # 业务编排：创建成功后发欢迎短信（异步，不阻塞响应）
        # await sms_service.send_welcome(user.phone)

        return user

    async def get_user(self, user_id: int) -> "User":
        user = await self.repo.find_by_id(user_id)
        if not user:
            raise AppException(ErrorCode.USER_NOT_FOUND, "用户不存在", 404)
        return user

    async def update_user(self, user_id: int, data) -> "User":
        user = await self.get_user(user_id)  # 复用上面的逻辑
        return await self.repo.update(user, data)

    async def delete_user(self, user_id: int):
        user = await self.get_user(user_id)
        # 业务规则：不能删除自己
        # 业务规则：管理员不能被删除
        await self.repo.delete(user)

    async def list_users(self, page: int, size: int, keyword: str = None):
        return await self.repo.find_all(page, size, keyword)

# modules/user/repository.py
from sqlalchemy import select, func, or_
from .model import User

class UserRepository:
    """用户仓库 —— 所有用户相关的数据库操作在这里"""

    def __init__(self, db: AsyncSession = Depends(get_db)):
        self.db = db

    async def find_by_id(self, user_id: int) -> User | None:
        result = await self.db.execute(select(User).where(User.id == user_id))
        return result.scalar_one_or_none()

    async def find_by_phone(self, phone: str) -> User | None:
        result = await self.db.execute(select(User).where(User.phone == phone))
        return result.scalar_one_or_none()

    async def find_by_username(self, username: str) -> User | None:
        result = await self.db.execute(select(User).where(User.username == username))
        return result.scalar_one_or_none()

    async def create(self, **kwargs) -> User:
        user = User(**kwargs)
        self.db.add(user)
        await self.db.commit()
        await self.db.refresh(user)
        return user

    async def update(self, user: User, data) -> User:
        for key, value in data.model_dump(exclude_unset=True).items():
            setattr(user, key, value)
        await self.db.commit()
        await self.db.refresh(user)
        return user

    async def delete(self, user: User):
        await self.db.delete(user)
        await self.db.commit()

    async def find_all(self, page: int, size: int, keyword: str = None) -> tuple[list[User], int]:
        query = select(User)
        count_query = select(func.count()).select_from(User)

        if keyword:
            filter_cond = or_(
                User.username.contains(keyword),
                User.phone.contains(keyword),
            )
            query = query.where(filter_cond)
            count_query = count_query.where(filter_cond)

        query = query.offset((page - 1) * size).limit(size)

        items = (await self.db.execute(query)).scalars().all()
        total = (await self.db.execute(count_query)).scalar()
        return items, total

// 一个 Model 上千行
class User extends Model {
    protected $table = 'users';           // 表结构定义
    public function orders() { ... }      // 关联关系
    public function scopeActive($q) { .. } // 查询作用域
    public function setPasswordAttr($v) { } // 修改器
    public function getAvatarAttr($v) { }  // 获取器
    public static function getByPhone($p) {} // 自定义查询
    // ... 还有 30 个方法

# modules/user/model.py
from sqlalchemy import Column, Integer, String, DateTime, Boolean
from core.database import Base

class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, autoincrement=True)
    username = Column(String(50), unique=True, nullable=False, comment="用户名")
    phone = Column(String(20), unique=True, nullable=False, comment="手机号")
    password = Column(String(255), nullable=False, comment="密码")
    email = Column(String(100), nullable=True, comment="邮箱")
    avatar = Column(String(255), nullable=True, comment="头像")
    is_active = Column(Boolean, default=True, comment="是否启用")
    created_at = Column(DateTime, server_default=func.now())
    updated_at = Column(DateTime, onupdate=func.now())

class 制茶师:
    def __init__(self):
        self.茶叶 = 福建茶园()    # 写死了！
        self.奶源 = 蒙牛牧场()    # 写死了！
        self.杯子 = 塑料杯厂()    # 写死了！

class 制茶师:
    def __init__(self, 茶叶, 奶源, 杯子):  # 从外部传进来
        self.茶叶 = 茶叶
        self.奶源 = 奶源
        self.杯子 = 杯子

# 生产时：用真实的供应商
制茶师(福建茶园(), 蒙牛牧场(), 塑料杯厂())

# 测试时：用假的供应商（不真的进货，不真的花钱）
制茶师(假茶园(), 假牧场(), 假杯厂())

class UserController {
    public function create() {
        $logic = new UserLogic();      // 硬编码依赖
        $validate = new UserValidate(); // 硬编码依赖
    }
}

正常情况（没有IoC）：  我要用车 → 我自己造一辆车 → 我开车
IoC之后：             我要用车 → 我告诉工厂"我要车" → 工厂给我一辆车 → 我开车

# ❌ 没有 IoC —— 制茶师自己控制供应商的创建
class 制茶师:
    def __init__(self):
        self.茶叶 = 福建茶园()   # "我自己找的！"
        # 控制权在制茶师手里

# ✅ 有 IoC —— 制茶师不控制，外部传进来
class 制茶师:
    def __init__(self, 茶叶):     # "你给我什么，我用什么"
        self.茶叶 = 茶叶
        # 控制权在外部调用者手里

# ❌ 没有 DIP —— 依赖具体品牌
class 制茶师:
    def __init__(self, 茶叶: 福建茶园):   # 只能用福建茶园！
        self.茶叶 = 茶叶

# 现在想换云南茶园？改类型声明，改代码。

# ✅ 有 DIP —— 依赖抽象（接口）
class 制茶师:
    def __init__(self, 茶叶: 茶叶供应商):  # 只要是"茶叶供应商"就行！
        self.茶叶 = 茶叶

# 福建茶园、云南茶园、假茶园，只要实现了"茶叶供应商"接口，都能传进来

没有 IoC，没有 DIP：
  制茶师 ──new──→ 福建茶园（具体类）
  硬编码，改不了，测不了

有 IoC，没有 DIP：
  制茶师 ←──传─── 福建茶园（具体类）
  能从外部传了，但类型写死了，只能传福建茶园

有 IoC + DIP：
  制茶师 ←──传─── 任何实现了"茶叶供应商"接口的类
  能从外部传，类型也灵活了，想传谁传谁

# ❌ 没有IoC：UserController 内部 new UserService，UserService 内部 new UserRepository
# 你想测 Controller 的逻辑，但它会真的连数据库——测试又慢又不稳定

# ✅ 有IoC+DIP：传入 FakeUserRepository（假仓库，不连数据库）
service = UserService(repo=FakeUserRepository())  # 秒完，不碰数据库

# ❌ 没有IoC：到处 new MySQLRepository()，换 PostgreSQL 要改50个文件

# ✅ 有IoC+DIP：只改一处注册
# 原来：app.dependency_overrides[UserRepository] = MySQLUserRepository
# 现在：app.dependency_overrides[UserRepository] = PGUserRepository
# 业务代码一行不改

# ❌ 没有IoC：看函数签名猜不到依赖
def create_user(data):
    logic = new UserLogic()  # 藏在方法内部，读代码才能发现
    ...

# ✅ 有IoC：依赖一目了然
def create_user(data, service: UserService = Depends()):  # 签名就告诉你：我需要 UserService
    ...

# 步骤1：定义依赖（我需要什么）
async def get_db():
    """数据库会话——请求进来时创建，请求结束时关闭"""
    async with AsyncSessionLocal() as session:
        yield session

class UserRepository:
    """声明：我需要一个数据库会话"""
    def __init__(self, db: AsyncSession = Depends(get_db)):
        self.db = db

class UserService:
    """声明：我需要一个仓库"""
    def __init__(self, repo: UserRepository = Depends()):
        self.repo = repo

# 步骤2：在路由中使用（框架自动注入整条依赖链）
@router.post("/users")
async def create_user(
    data: CreateUserRequest,
    service: UserService = Depends(),        # ← 自动创建并注入
):
    return ApiResponse.success(await service.create_user(data))

1. 看到 Depends(UserService)
2. 发现 UserService 需要 UserRepository
3. 发现 UserRepository 需要 AsyncSession
4. 自动按顺序创建：
   AsyncSession → UserRepository → UserService
5. 注入到路由函数
6. 请求结束后自动关闭 AsyncSession

# 单元测试——不需要数据库
async def test_create_user():
    fake_repo = FakeUserRepository()         # 假仓库，不连数据库
    service = UserService(repo=fake_repo)    # 手动注入假仓库
    user = await service.create_user(...)
    assert user.username == "张三"

// 插件系统——注册依赖到 fastify 实例
const fp = require('fastify-plugin')

async function dbPlugin(fastify, opts) {
  const db = await createDB(opts.database)
  fastify.decorate('db', db)
}

async function userPlugin(fastify, opts) {
  const repo = new UserRepository(fastify.db)
  const service = new UserService(repo)
  fastify.decorate('userRepository', repo)
  fastify.decorate('userService', service)
}

// 注册
fastify.register(fp(dbPlugin), { database: config.db })
fastify.register(fp(userPlugin))

// 路由中使用
fastify.get('/users/me', async (request, reply) => {
  return await fastify.userService.getUser(request.currentUser.id)
})

// 测试时：注册假的插件
fastify.register(fp(async (f) => {
  f.decorate('userRepository', new FakeUserRepo())
  f.decorate('userService', new UserService(f.userRepository))
}))

收银员：您好，点什么？
顾客：我要一杯 -3°C 的奶茶
收银员：等等，温度不能是负数...让我翻手册查规则...
       手册第 37 页写着"温度范围 0-100"...好的，不行。

顾客进门 → 安检门自动扫描 → 温度-3°C？嘟！禁止入内！
收银员：你好，点什么？（只管接单，不管校验）

class UserValidate extends Validate {
    protected $rule = [
        'username' => 'require|max:50',
        'phone'    => 'require|regex:^1[3-9]\d{9}$|unique:user',
        'password' => 'require|min:6|max:20',
    ];
    protected $message = [
        'username.require' => '用户名必填',
        'phone.regex'      => '手机号格式不正确',
    ];
}

# modules/user/schema.py
from pydantic import BaseModel, Field, EmailStr, field_validator
import re

class CreateUserRequest(BaseModel):
    """声明即验证——每个字段的类型就是它的验证规则"""

    username: str = Field(
        ...,                    # ... = 必填
        min_length=2,           # 最短 2 字符
        max_length=50,          # 最长 50 字符
        description="用户名"
    )

    phone: str = Field(
        ...,
        pattern=r'^1[3-9]\d{9}$',  # 正则校验
        description="手机号"
    )

    password: str = Field(
        ...,
        min_length=6,
        max_length=20,
        description="密码"
    )

    email: EmailStr | None = Field(None, description="邮箱")  # EmailStr 自动验证邮箱格式
    age: int = Field(ge=0, le=150, description="年龄")         # ge=大于等于, le=小于等于

    # 复杂业务校验用 field_validator
    @field_validator('password')
    @classmethod
    def password_must_have_letter_and_digit(cls, v: str) -> str:
        if not re.search(r'[A-Za-z]', v) or not re.search(r'[0-9]', v):
            raise ValueError('密码必须同时包含字母和数字')
        return v

@router.post("/users")
async def create_user(data: CreateUserRequest):
    # data.phone 一定是合法手机号
    # data.age 一定在 0-150 之间
    # data.email 一定是合法邮箱（如果有的话）
    # 不需要写一行校验代码！
    return await user_service.create_user(data)

{
  "detail": [
    {"loc": ["body", "phone"], "msg": "String should match pattern '^1[3-9]\\d{9}$'"},
    {"loc": ["body", "age"], "msg": "Input should be less than or equal to 150"}
  ]
}

const createUserSchema = {
  type: 'object',
  required: ['username', 'phone', 'password'],
  properties: {
    username: { type: 'string', minLength: 2, maxLength: 50 },
    phone: { type: 'string', pattern: '^1[3-9]\\d{9}$' },
    password: { type: 'string', minLength: 6, maxLength: 20 },
    email: { type: 'string', format: 'email' },
    age: { type: 'integer', minimum: 0, maximum: 150 }
  }
}

fastify.post('/users', { schema: { body: createUserSchema } }, async (req) => {
  // req.body 已验证
  return await fastify.userService.createUser(req.body)
})

✅ "好的，珍珠奶茶已下单，订单号 #042，等 5 分钟"
❌ "抱歉，芒果今天卖完了（错误码：3001）"

"042，奶茶，5分钟"   ← 信息不全
"没了"               ← 什么没了？

// 到处直接 return，没有异常体系
return json(['code' => 0, 'msg' => '用户不存在']);
return json(['code' => -1, 'msg' => '参数错误']);
return json(['code' => 400, 'msg' => '没有权限']);

# common/response.py
from typing import Generic, TypeVar
from pydantic import BaseModel

T = TypeVar("T")

class ApiResponse(BaseModel, Generic[T]):
    """统一响应格式——前端只认这一种"""
    code: int = 1
    msg: str = "操作成功"
    data: T | None = None

    @classmethod
    def success(cls, data: T = None, msg: str = "操作成功"):
        return cls(code=1, msg=msg, data=data)

    @classmethod
    def fail(cls, msg: str = "操作失败", code: int = 0):
        return cls(code=code, msg=msg, data=None)

class PageData(BaseModel, Generic[T]):
    """分页数据"""
    items: list[T]
    total: int
    page: int
    size: int

# common/exceptions.py
from enum import IntEnum

class ErrorCode(IntEnum):
    """
    统一错误码——每个错误有唯一编号，按模块分段
    前端拿到 code 就知道具体什么错误，不用匹配字符串
    """
    # 通用错误 1000-1999
    SUCCESS = 0
    UNKNOWN = 1000
    PARAM_ERROR = 1001
    NOT_FOUND = 1002

    # 用户模块 2000-2999
    USER_NOT_FOUND = 2000
    USER_PHONE_EXISTS = 2001
    USER_PASSWORD_ERROR = 2002
    USER_DISABLED = 2003
    USERNAME_EXISTS = 2004

    # 订单模块 3000-3999
    ORDER_NOT_FOUND = 3000
    ORDER_STOCK_INSUFFICIENT = 3001
    ORDER_STATUS_ERROR = 3002

    # 权限模块 4000-4999
    UNAUTHORIZED = 4000
    FORBIDDEN = 4001
    TOKEN_EXPIRED = 4002

class AppException(Exception):
    """应用异常——业务代码只管 raise，不管怎么返回"""
    def __init__(self, code: ErrorCode, msg: str = None, http_status: int = 400):
        self.code = code
        self.msg = msg or code.name
        self.http_status = http_status

# 常用异常快捷方式
class NotFoundException(AppException):
    def __init__(self, msg: str = "资源不存在"):
        super().__init__(ErrorCode.NOT_FOUND, msg, 404)

class UnauthorizedException(AppException):
    def __init__(self, msg: str = "未登录"):
        super().__init__(ErrorCode.UNAUTHORIZED, msg, 401)

class ForbiddenException(AppException):
    def __init__(self, msg: str = "无权限"):
        super().__init__(ErrorCode.FORBIDDEN, msg, 403)

# common/exception_handler.py
from fastapi import Request
from fastapi.responses import JSONResponse

async def app_exception_handler(request: Request, exc: AppException):
    """统一处理业务异常"""
    return JSONResponse(
        status_code=exc.http_status,
        content={"code": exc.code, "msg": exc.msg, "data": None}
    )

async def validation_exception_handler(request: Request, exc: ValidationError):
    """统一处理 Pydantic 验证异常"""
    errors = exc.errors()
    first_error = errors[0]["msg"] if errors else "参数错误"
    return JSONResponse(
        status_code=422,
        content={"code": 1001, "msg": f"参数校验失败: {first_error}", "data": None}
    )

async def global_exception_handler(request: Request, exc: Exception):
    """兜底——处理所有未预期的异常"""
    logger.error(f"未预期错误: {exc}", exc_info=True)
    return JSONResponse(
        status_code=500,
        content={"code": 1000, "msg": "服务器内部错误", "data": None}
    )

# main.py 注册
app.add_exception_handler(AppException, app_exception_handler)
app.add_exception_handler(ValidationError, validation_exception_handler)
app.add_exception_handler(Exception, global_exception_handler)

# Service 中只管 raise——干净利落
async def login(self, phone: str, password: str):
    user = await self.repo.find_by_phone(phone)
    if not user:
        raise AppException(ErrorCode.USER_NOT_FOUND, "用户不存在")
    if not CryptoUtils.verify_password(password, user.password):
        raise AppException(ErrorCode.USER_PASSWORD_ERROR, "密码错误")
    if not user.is_active:
        raise AppException(ErrorCode.USER_DISABLED, "账号已停用")
    # ... 生成 token

// 前端全局拦截器
if (res.code !== 1) {
  switch (res.code) {
    case 2001: showToast('该手机号已注册，请直接登录'); break
    case 2002: showToast('密码错误，还剩 3 次机会'); break
    case 4002: redirectTo('/login'); break   // 登录过期，跳转登录页
    default:   showToast(res.msg)
  }
}

// common/exceptions.js
class AppException extends Error {
  constructor(code, msg, httpStatus = 400) {
    super(msg)
    this.code = code
    this.msg = msg
    this.httpStatus = httpStatus
  }
}

// app.js 全局错误处理
fastify.setErrorHandler(async (error, request, reply) => {
  if (error instanceof AppException) {
    return reply.status(error.httpStatus).send({
      code: error.code, msg: error.msg, data: null
    })
  }
  if (error.validation) {
    return reply.status(422).send({
      code: 1001, msg: `参数校验失败: ${error.message}`, data: null
    })
  }
  request.log.error(error)
  return reply.status(500).send({
    code: 1000, msg: '服务器内部错误', data: null
  })
})

门口 → 测温 → 扫码登记 → 洗手消毒 → 点单台
        ↓        ↓          ↓
     发烧劝退  记录轨迹   卫生保障

# modules/permission/service.py
from enum import Enum

class DataScope(Enum):
    ALL = "all"        # 超级管理员：看全部
    DEPT = "dept"      # 部门经理：看本部门
    SELF = "self"      # 普通员工：仅自己

class PermissionService:
    async def check_permission(self, user_id: int, perm_code: str) -> bool:
        """操作权限检查"""
        perms = await self._get_cached_permissions(user_id)
        return perm_code in perms

    async def get_data_scope(self, user_id: int) -> DataScope:
        """数据权限范围——likeadmin 缺失的能力"""
        roles = await self.repo.get_user_roles(user_id)
        if "admin" in roles: return DataScope.ALL
        if "manager" in roles: return DataScope.DEPT
        return DataScope.SELF

# core/dependencies.py
from fastapi import Depends

async def get_current_user(
    authorization: str = Header(...),
    auth_service: AuthService = Depends()
):
    """认证依赖——验证登录态"""
    user = await auth_service.verify_token(authorization)
    if not user:
        raise UnauthorizedException("登录已过期")
    return user

class PermissionChecker:
    """权限依赖——声明式，精确到每个接口"""
    def __init__(self, perm_code: str):
        self.perm_code = perm_code

    async def __call__(
        self,
        current_user = Depends(get_current_user),   # 先认证
        perm_service: PermissionService = Depends()
    ):
        if not await perm_service.check_permission(current_user.id, self.perm_code):
            raise ForbiddenException(f"缺少权限: {self.perm_code}")
        return current_user

# 路由中使用——每个接口精确指定需要什么权限
@router.get("/users", dependencies=[Depends(PermissionChecker("user:list"))])
@router.post("/users", dependencies=[Depends(PermissionChecker("user:create"))])
@router.delete("/users/{id}", dependencies=[Depends(PermissionChecker("user:delete"))])

// 认证中间件 —— 为什么要写成独立函数而不是写在路由里？
// 原因1：复用——几十个路由都需要认证，不用每个都写一遍
// 原因2：可测试——可以单独测试认证逻辑
// 原因3：可替换——换成 OAuth 只需改这一个函数
async function authMiddleware(request, reply) {
  const token = request.headers.authorization?.replace('Bearer ', '')
  if (!token) {
    throw new AppException(ErrorCode.UNAUTHORIZED, '未登录', 401)
  }
  try {
    const decoded = await fastify.jwt.verify(token)    // 用 @fastify/jwt 验证
    request.currentUser = await fastify.userService.getUser(decoded.userId)
  } catch (err) {
    throw new AppException(ErrorCode.TOKEN_EXPIRED, '登录已过期', 401)
  }
}

// 权限中间件工厂 —— 为什么是"工厂"？
// 因为每个接口需要的权限不同，requirePermission('user:list') 返回一个
// 专门检查 user:list 权限的中间件函数。这比 if/else 灵活得多。
function requirePermission(code) {
  return async function checkPermission(request, reply) {
    const userId = request.currentUser.id
    const permissions = await fastify.permService.getPermissions(userId)
    if (!permissions.includes(code)) {
      throw new AppException(ErrorCode.FORBIDDEN, `缺少权限: ${code}`, 403)
    }
  }
}

// 路由中使用 —— 注意 preHandler 数组的顺序：
// 先认证（拿到用户信息），再检查权限（根据用户信息判断）
// 顺序不能反，否则权限中间件拿不到 currentUser
fastify.get('/users', {
  preHandler: [authMiddleware, requirePermission('user:list')]
}, async (request, reply) => {
  const { page = 1, size = 10, keyword } = request.query
  const { items, total } = await fastify.userService.listUsers(page, size, keyword)
  return ApiResponse.success(ApiResponse.pageData(items, total, page, size))
})

fastify.post('/users', {
  preHandler: [authMiddleware, requirePermission('user:create')],
  schema: { body: createUserSchema }
}, async (request, reply) => {
  const user = await fastify.userService.createUser(request.body)
  return ApiResponse.success(user, '创建成功')
})

fastify.delete('/users/:userId', {
  preHandler: [authMiddleware, requirePermission('user:delete')]
}, async (request, reply) => {
  await fastify.userService.deleteUser(request.params.userId)
  return ApiResponse.success(null, '删除成功')
})

# modules/permission/repository.py
class PermissionRepository:
    async def get_user_roles(self, user_id: int) -> list[str]:
        """获取用户的角色列表"""
        result = await self.db.execute(
            select(Role.code)
            .join(user_roles, Role.id == user_roles.c.role_id)
            .where(user_roles.c.user_id == user_id)
        )
        return [row[0] for row in result.all()]

# modules/permission/service.py
class PermissionService:
    async def get_data_scope(self, user_id: int) -> DataScope:
        """
        获取数据权限范围。
        为什么不直接在 Service/Repository 里写 if/else？
        因为数据权限是"横切关注点"——它不是某个业务的逻辑，
        而是所有涉及数据查询的地方都需要的能力。
        把它抽成独立服务，任何 Repository 都能调用，不用重复写。
        """
        roles = await self.repo.get_user_roles(user_id)
        if "admin" in roles:
            return DataScope.ALL       # 管理员：看全部数据
        if "manager" in roles:
            return DataScope.DEPT      # 部门经理：看本部门数据
        return DataScope.SELF          # 普通员工：只看自己的数据

# modules/order/repository.py —— 数据权限在这里自动过滤
class OrderRepository:
    async def find_all(self, user_id: int, page: int, size: int):
        """
        为什么把数据权限过滤放在 Repository 而不是 Service？
        因为 Repository 是"唯一直接操作数据库的地方"。
        如果在 Service 里过滤，万一有人绕过 Service 直接调 Repository，
        数据权限就形同虚设。放在 Repository 里，从根源保证安全。
        """
        query = select(Order)

        # 根据数据权限自动添加过滤条件
        scope = await self.perm_service.get_data_scope(user_id)
        if scope == DataScope.SELF:
            # 只看自己创建的订单
            query = query.where(Order.creator_id == user_id)
        elif scope == DataScope.DEPT:
            # 只看本部门的订单
            dept_id = await self.perm_service.get_user_dept_id(user_id)
            query = query.where(Order.dept_id == dept_id)
        # DataScope.ALL → 不加过滤，看全部

        # 无论如何，查询的后续逻辑（分页、排序）不受影响
        count_query = select(func.count()).select_from(query.subquery())
        total = (await self.db.execute(count_query)).scalar()

        items = (await self.db.execute(
            query.offset((page - 1) * size).limit(size)
        )).scalars().all()

        return items, total

配方（代码）：珍珠奶茶 = 红茶底 + 奶精 + 珍珠
  → 全国统一，写死在流程里

运营参数（配置）：糖度 70%、珍珠 50g/份、营业时间 9-22 点
  → 每家店不同，放在店长手册里随时调

# core/config.py
from pydantic_settings import BaseSettings
from functools import lru_cache

class Settings(BaseSettings):
    """
    应用配置——自动从 .env 读取，有类型校验。
    
    为什么用 pydantic-settings 而不是直接 os.getenv()？
    1. 类型安全：DB_PORT 一定是 int，不会出现 "3306" 字符串当数字用
    2. 启动校验：必填项缺失时，应用启动就报错，而不是运行到一半才崩
    3. IDE 友好：settings.DB_PORT 有自动补全，os.getenv("DB_PORT") 没有
    4. 默认值集中：所有默认值在 Settings 类里一目了然
    """

    # 应用
    APP_NAME: str = "通用后台"
    APP_VERSION: str = "1.0.0"
    DEBUG: bool = False

    # 数据库
    DB_HOST: str = "localhost"
    DB_PORT: int = 3306          # pydantic 自动把 "3306" 转成 int
    DB_USER: str = "root"
    DB_PASSWORD: str = ""        # 敏感信息从 .env 读取，不写死在代码里
    DB_NAME: str = "admin"

    # Redis
    REDIS_URL: str = "redis://localhost:6379/0"

    # JWT
    JWT_SECRET: str = "change-me-in-production"  # 生产环境必须通过 .env 覆盖！
    JWT_ALGORITHM: str = "HS256"
    JWT_EXPIRE_MINUTES: int = 1440  # 24 小时

    # 上传
    UPLOAD_DIR: str = "./uploads"
    MAX_UPLOAD_SIZE: int = 10 * 1024 * 1024  # 10MB

    # 为什么用 @property 而不是直接写 DB_URL 字段？
    # 因为 database_url 是从其他字段"拼出来"的，不是独立的配置项。
    # 如果写成字段，用户需要在 .env 里同时配置 DB_HOST 和 DB_URL，
    # 容易出现不一致。用 @property 保证永远一致。
    @property
    def database_url(self) -> str:
        return f"mysql+aiomysql://{self.DB_USER}:{self.DB_PASSWORD}@{self.DB_HOST}:{self.DB_PORT}/{self.DB_NAME}"

    model_config = {"env_file": ".env", "case_sensitive": True}

# 为什么用 @lru_cache()？
# 因为 Settings() 每次调用都会读 .env 文件、解析类型、校验数据。
# 整个应用只需要加载一次，后续调用直接返回缓存的结果。
# 这比"全局变量"优雅，因为不会在模块导入时就执行（避免循环依赖）。
@lru_cache()
def get_settings() -> Settings:
    return Settings()

# .env —— 开发环境配置（不提交 Git！）
APP_NAME=通用后台-开发环境
DEBUG=true

DB_HOST=localhost
DB_PORT=3306
DB_USER=root
DB_PASSWORD=dev_password_123
DB_NAME=admin_dev

REDIS_URL=redis://localhost:6379/0

JWT_SECRET=dev-secret-key-do-not-use-in-production
JWT_EXPIRE_MINUTES=10080   # 开发环境7天，方便调试

UPLOAD_DIR=./uploads
MAX_UPLOAD_SIZE=10485760

# .env.example —— 配置模板（提交 Git，告诉别人需要哪些配置）
# 复制此文件为 .env，然后填写实际值
APP_NAME=
DEBUG=false

DB_HOST=
DB_PORT=3306
DB_USER=
DB_PASSWORD=
DB_NAME=

REDIS_URL=

JWT_SECRET=             # 必填！生产环境请使用强密钥
JWT_EXPIRE_MINUTES=1440

UPLOAD_DIR=./uploads
MAX_UPLOAD_SIZE=10485760

.env          → 包含真实密码，绝不能提交（类似店长的保险箱密码）
.env.example  → 只有配置项名称和默认值，可以提交（类似"需要填哪些表"的清单）

新人入职流程：
1. git clone 项目
2. cp .env.example .env     → 复制模板
3. 填写真实密码              → 生成自己的 .env
4. 启动项目                  → pydantic-settings 自动读取 .env

大抽屉 common.php：
  剪刀、计算器、温度计、封口机、量杯、计时器、订书机...

string_utils.py：  字符串相关（截断、脱敏、随机串）
date_utils.py：     日期相关（格式化、天数差）
crypto_utils.py：   加密相关（哈希、JWT）
file_utils.py：     文件相关（上传、压缩）

# common/utils/string_utils.py
class StringUtils:
    """
    字符串工具类。
    
    为什么用类而不是直接写函数？
    1. 命名空间隔离：StringUtils.mask_phone() 不会和其他模块的 mask_phone 冲突
    2. 语义清晰：看到 StringUtils 就知道里面是字符串相关的
    3. 可扩展：以后加实例方法（如需要状态的工具）不用改调用方式
    """
    @staticmethod
    def mask_phone(phone: str) -> str:
        """手机号脱敏：138****1234"""
        return phone[:3] + "****" + phone[-4:]

    @staticmethod
    def random_code(length: int = 6) -> str:
        """生成随机数字验证码"""
        return ''.join(random.choices('0123456789', k=length))

    @staticmethod
    def truncate(text: str, max_length: int = 50, suffix: str = "...") -> str:
        """截断长文本，保留前 max_length 个字符"""
        if len(text) <= max_length:
            return text
        return text[:max_length - len(suffix)] + suffix

# common/utils/crypto_utils.py
class CryptoUtils:
    """
    加密工具类。
    为什么单独放一个文件而不是和 string_utils 混在一起？
    因为加密工具的依赖（bcrypt、jose）和其他工具完全不同。
    如果只用到字符串工具的项目，不需要安装加密依赖。
    这就是"按职责拆分"的好处——各取所需，不连带安装没用的包。
    """
    _pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

    @staticmethod
    def hash_password(password: str) -> str:
        """密码加密——为什么用 bcrypt 而不是 MD5/SHA256？
        因为 MD5/SHA256 是"哈希"不是"加密"，可以被彩虹表破解。
        bcrypt 自带盐值（salt），同一密码每次加密结果不同，更安全。"""
        return CryptoUtils._pwd_context.hash(password)

    @staticmethod
    def verify_password(plain: str, hashed: str) -> bool:
        """密码校验"""
        return CryptoUtils._pwd_context.verify(plain, hashed)

    @staticmethod
    def create_token(data: dict, secret: str, expires_minutes: int = 1440) -> str:
        """生成 JWT Token——为什么用 JWT 而不是 Session？
        因为 JWT 是无状态的，服务端不需要存储会话信息。
        多台服务器之间不用共享 Session，天然支持水平扩展。"""
        to_encode = data.copy()
        expire = datetime.utcnow() + timedelta(minutes=expires_minutes)
        to_encode.update({"exp": expire})
        return jwt.encode(to_encode, secret, algorithm="HS256")

    @staticmethod
    def verify_token(token: str, secret: str) -> dict:
        """验证 JWT Token"""
        return jwt.decode(token, secret, algorithms=["HS256"])

# common/utils/date_utils.py
class DateUtils:
    """日期工具类"""
    @staticmethod
    def now_beijing() -> datetime:
        """获取北京时间——为什么单独写？
        因为服务器可能部署在不同时区，业务上统一用北京时间。"""
        return datetime.now(timezone(timedelta(hours=8)))

    @staticmethod
    def format_ago(dt: datetime) -> str:
        """人性化时间：刚刚、5分钟前、3小时前、昨天"""
        delta = datetime.now() - dt
        seconds = delta.total_seconds()
        if seconds < 60: return "刚刚"
        if seconds < 3600: return f"{int(seconds // 60)}分钟前"
        if seconds < 86400: return f"{int(seconds // 3600)}小时前"
        if seconds < 172800: return "昨天"
        return f"{int(seconds // 86400)}天前"

// common/utils/stringUtils.js
class StringUtils {
  static maskPhone(phone) {
    // 手机号脱敏：138****1234
    return phone.slice(0, 3) + '****' + phone.slice(-4)
  }

  static randomCode(length = 6) {
    // 生成随机数字验证码
    return Array.from({ length }, () => Math.floor(Math.random() * 10)).join('')
  }

  static truncate(text, maxLength = 50, suffix = '...') {
    return text.length <= maxLength ? text : text.slice(0, maxLength - suffix.length) + suffix
  }
}

// common/utils/cryptoUtils.js
const bcrypt = require('bcrypt')
const jwt = require('jsonwebtoken')

class CryptoUtils {
  static async hashPassword(password) {
    // bcrypt 自动加盐，同一个密码每次 hash 结果不同
    return await bcrypt.hash(password, 10)
  }

  static async verifyPassword(plain, hashed) {
    return await bcrypt.compare(plain, hashed)
  }

  static createToken(data, secret, expiresMinutes = 1440) {
    return jwt.sign(data, secret, { expiresIn: expiresMinutes + 'm' })
  }

  static verifyToken(token, secret) {
    return jwt.verify(token, secret)
  }
}

// 使用时按需引入，不用全部加载
const { StringUtils } = require('./common/utils/stringUtils')
const { CryptoUtils } = require('./common/utils/cryptoUtils')

// 在 Service 中使用
class UserService {
  async createUser(data) {
    const hashedPwd = await CryptoUtils.hashPassword(data.password)
    const user = await this.repo.create({ ...data, password: hashedPwd })
    return user
  }
}