FastAPI注册端点数据校验方案咨询及代码优化建议请求

当前的校验逻辑功能上可行，但存在明显缺陷：大量冗余的if判断导致代码可读性差、维护成本高，且没有充分利用框架和数据库的原生能力，属于重复造轮子；同时提前查询数据库检查phone/login唯一性的做法，会增加不必要的数据库请求，还可能出现并发场景下的竞态问题（比如两个请求同时检查同一手机号未被注册，随后都写入数据库，违反唯一约束）。

以下基于FastAPI+SQLAlchemy的技术栈给出优化方案（适配你代码中的技术特征）：

1. 用Pydantic模型接管基础校验 #

将必填字段校验、格式校验直接封装到UserIn模型中，利用Pydantic的原生能力自动完成校验，无需在接口中写if判断：

from pydantic import BaseModel, Field, validator
import re

class UserIn(BaseModel):
    # 用...标记必填字段，Pydantic会自动校验非空
    name: str = Field(..., description="姓名（必填）")
    secondname: str = Field(..., description="中间名（必填）")
    lastname: str = Field(..., description="姓氏（必填）")
    phone: str = Field(..., description="手机号（必填）")
    login: str = Field(..., description="登录名（必填）")
    password: str = Field(..., min_length=6, description="密码（至少6位）")

    # 自定义手机号格式校验器
    @validator('phone')
    def check_phone_format(cls, value):
        # 可根据实际需求调整正则规则
        if not re.match(r'^1[3-9]\d{9}$', value):
            raise ValueError("请输入正确的手机号格式")
        return value

FastAPI会自动对请求参数做校验，不符合规则时直接返回400错误，且提示信息对应具体的校验失败项。

2. 利用数据库唯一约束处理重复值，减少冗余查询 #

直接尝试写入数据库，捕获唯一约束冲突的异常，替代提前查询的逻辑：

from sqlalchemy.exc import IntegrityError

@post("/register")
async def register(user: UserIn):
    hashed_password = HashedData.hash_password(user.password)
    try:
        # 直接调用注册逻辑，无需提前查询phone/login是否存在
        success = RegisterCRUD.register(
            user.name, user.secondname, user.lastname,
            user.phone, user.login, hashed_password
        )
        if not success:
            raise HTTPException(status_code=501, detail="用户创建失败")
        return JSONResponse(status_code=201, content={"message": "用户创建成功"})
    except IntegrityError as e:
        # 解析数据库异常信息，判断冲突字段
        error_detail = str(e.orig)
        if "phone" in error_detail:
            raise HTTPException(status_code=400, detail="该手机号已注册")
        elif "login" in error_detail:
            raise HTTPException(status_code=400, detail="该登录名已被占用")
        # 其他完整性错误（如外键冲突）统一处理
        raise HTTPException(status_code=400, detail="数据校验失败")

这种方式既减少了两次数据库查询，又能通过数据库约束保证数据一致性，还能避免并发场景下的竞态问题。

3. 代码解耦：分离控制器与业务逻辑 #

当前控制器继承RegisterCRUD的做法耦合度太高，建议拆分出服务层，让控制器只负责请求接收和响应返回：

# 新增服务层
class AuthService:
    @staticmethod
    async def handle_user_register(user_in: UserIn):
        hashed_pwd = HashedData.hash_password(user_in.password)
        try:
            await RegisterCRUD.create(
                name=user_in.name,
                secondname=user_in.secondname,
                lastname=user_in.lastname,
                phone=user_in.phone,
                login=user_in.login,
                password=hashed_pwd
            )
        except IntegrityError as e:
            error_detail = str(e.orig)
            if "phone" in error_detail:
                raise HTTPException(status_code=400, detail="该手机号已注册")
            elif "login" in error_detail:
                raise HTTPException(status_code=400, detail="该登录名已被占用")
            raise HTTPException(status_code=400, detail="数据错误")
        return {"message": "用户创建成功"}

# 控制器简化
class AuthenticateController(Controller):
    prefix = "/api/auth"
    tags = ["auth"]
    refresh_token_ttl = 30*60
    access_token_ttl = 15*60

    @post("/register")
    async def register(user: UserIn):
        result = await AuthService.handle_user_register(user)
        return JSONResponse(status_code=201, content=result)

拆分后代码职责更清晰，后续修改业务逻辑或CRUD操作时，不会影响控制器层。

• 消除冗余if判断，代码更简洁易读
• 利用框架和数据库原生能力，减少重复代码，提升性能
• 避免并发场景下的数据一致性问题
• 代码分层清晰，维护成本降低

内容的提问来源于stack exchange，提问作者Genry