FastAPI中CRUD层字段验证的正确方式探讨

我正在开发一个小型FastAPI披萨配送项目，需要在创建新对象前完成数据库级验证，比如确认披萨订单中的披萨名称是否在数据库中存在。

我注意到部分开发者直接在Pydantic Schema层完成这类验证，但认为该方式不合理——Schema应仅承担结构定义职责，不应直接操作数据库（职责过载）。为此自行设计了一套方案，将数据库验证分离为独立层，具体实现如下：

验证器抽象类+装饰器实现 #

class Validator(Protocol):
    """Validator abstract class"""
    def validation(db: Session):
        ...

def validator(validator: Validator):
    """Decorator for dynamic validator allocation"""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            try:
                db: Session = args[0]
                schema: schemas = args[1]
            except IndexError:
                raise AttributeError("Insufficient amount of arguments for validator")
            exception = validator(schema).validation(db)
            if exception: raise exception
            return func(*args, **kwargs)
        return wrapper
    return decorator

验证器实现类 #

class PizzaExistsValidator:
    def __init__(self, order: schemas.Order):
        self.order = order
    def validation(self, db: Session):
        if not all(get_pizza(db, ordered_pizza.pizza.name) != None
                   for ordered_pizza in self.order.pizzas):
            return PizzaExistenceException("Pizza does not exist")

CRUD层使用示例 #

@validator(PizzaExistsValidator)
def create_order(db: Session, order: schemas.Order):
    total_price = order.price
    order.pizzas = jsonable_encoder(order.pizzas)
    db_order = models.Order(**dict(order), price=total_price)
    db.add(db_order)
    db.commit()
    db.refresh(db_order)
    return db_order

我刚接触FastAPI，不清楚是否有更优实现方式，希望获取相关建议与反馈。

优化建议与反馈 #

1. 解决装饰器参数依赖硬编码问题
   当前装饰器通过args[0]和args[1]获取db和schema，一旦CRUD函数参数顺序变化就会出错。可以改用参数签名绑定的方式，按参数名获取值：

   import inspect
   
   def validator(validator_cls: Validator, db_param: str = "db", schema_param: str = "schema"):
       def decorator(func):
           @wraps(func)
           def wrapper(*args, **kwargs):
               sig = inspect.signature(func)
               bound_args = sig.bind(*args, **kwargs).arguments
               db = bound_args.get(db_param)
               schema = bound_args.get(schema_param)
               
               if not isinstance(db, Session):
                   raise AttributeError(f"Parameter '{db_param}' must be a SQLAlchemy Session")
               if not schema:
                   raise AttributeError(f"Parameter '{schema_param}' not found in function arguments")
               
               exception = validator_cls(schema).validation(db)
               if exception:
                   raise exception
               return func(*args, **kwargs)
           return wrapper
       return decorator
   

   使用时可灵活指定参数名：@validator(PizzaExistsValidator, schema_param="order")

2. 简化验证器异常逻辑
   当前验证器返回异常实例再由装饰器判断抛出，可改成验证器直接抛出异常，逻辑更直观：

   class PizzaExistsValidator:
       def __init__(self, order: schemas.Order):
           self.order = order
       def validation(self, db: Session):
           if not all(get_pizza(db, ordered_pizza.pizza.name) is not None
                      for ordered_pizza in self.order.pizzas):
               raise PizzaExistenceException("Pizza does not exist")
   

   装饰器中直接调用validator(schema).validation(db)即可，无需判断返回值。

3. 优化数据库查询性能
   当前循环查询每个披萨的方式效率低，可改为一次查询所有需要验证的披萨名称：

   def validation(self, db: Session):
       pizza_names = {ordered_pizza.pizza.name for ordered_pizza in self.order.pizzas}
       existing_names = {p.name for p in db.query(models.Pizza).filter(models.Pizza.name.in_(pizza_names)).all()}
       missing_names = pizza_names - existing_names
       if missing_names:
           raise PizzaExistenceException(f"Pizzas do not exist: {', '.join(missing_names)}")
   

   仅需一次数据库查询，还能明确告知用户缺失的披萨名称。

4. 提升验证器复用性
   可抽象通用存在性验证器，避免重复编写类似逻辑：

   class ExistenceValidator(Validator):
       def __init__(self, schema, model, field_name, schema_field_path):
           self.schema = schema
           self.model = model
           self.field_name = field_name
           self.schema_field_path = schema_field_path  # 例如"pizzas__pizza__name"
   
       def validation(self, db: Session):
           def get_nested_value(obj, path):
               parts = path.split("__")
               for part in parts:
                   if isinstance(obj, list):
                       obj = [getattr(item, part) for item in obj]
                   else:
                       obj = getattr(obj, part)
               return obj
   
           values = get_nested_value(self.schema, self.schema_field_path)
           values_set = set(values) if isinstance(values, list) else {values}
   
           existing_values = {getattr(p, self.field_name) for p in db.query(self.model).filter(getattr(self.model, self.field_name).in_(values_set)).all()}
           missing_values = values_set - existing_values
           if missing_values:
               raise ValueError(f"{self.model.__name__} with {self.field_name} {', '.join(missing_values)} do not exist")
   

   使用时无需编写新验证器：

   @validator(ExistenceValidator(schema=schemas.Order, model=models.Pizza, field_name="name", schema_field_path="pizzas__pizza__name"))
   def create_order(db: Session, order: schemas.Order):
       # 原有CRUD逻辑
   

5. 适配FastAPI依赖注入风格
   除装饰器外，可将验证逻辑封装为依赖项，更贴合FastAPI设计：

   def validate_order_pizzas(db: Session = Depends(get_db), order: schemas.Order = Depends()):
       pizza_names = {op.pizza.name for op in order.pizzas}
       existing_names = {p.name for p in db.query(models.Pizza).filter(models.Pizza.name.in_(pizza_names)).all()}
       missing = pizza_names - existing_names
       if missing:
           raise HTTPException(status_code=400, detail=f"Pizzas not found: {', '.join(missing)}")
       return order
   

   路由层组装依赖：

   @router.post("/orders/")
   def create_order_route(order: schemas.Order = Depends(validate_order_pizzas), db: Session = Depends(get_db)):
       return crud.create_order(db, order)
   

   这种方式将验证与CRUD完全解耦，路由层负责逻辑组装，CRUD层专注数据操作。

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