如何在FastAPI中通过Pydantic模型显式设置值为None?
问题:Repository模式下区分“未设置”和“显式设为None”的数据库更新方案
问题背景
我采用Repository模式更新数据库,当前逻辑仅更新值不为None的字段:
db_entry = get_entry(...) for name, value in entry_to_update.dict().items(): setattr(db_entry, name, value) if value else None
但现在需要支持显式将某个字段重置为None,当前逻辑无法实现——因为Pydantic的Optional字段如果未传入值,实例化后会自动填充为None,导致误更新:
class EntryToUpdateSchema(BaseModel): value_a: Optional[str] value_b: Optional[str] entry_to_update = EntryToUpdateSchema(value_a="A") # 实际得到 EntryToUpdateSchema(value_a="A", value_b=None) # 这会误把value_b设为None,但我本来只想忽略该字段
最优解决方案
方案1:使用Pydantic的exclude_unset参数(推荐)
Pydantic的dict()方法自带exclude_unset=True参数,只会返回实例化时显式传入的字段(无论值是否为None):
# 显式传入value_b=None时 entry_to_update = EntryToUpdateSchema(value_a="A", value_b=None) update_data = entry_to_update.dict(exclude_unset=True) # update_data 结果:{"value_a": "A", "value_b": None} # 仅传入value_a时 entry_to_update = EntryToUpdateSchema(value_a="A") update_data = entry_to_update.dict(exclude_unset=True) # update_data 结果:{"value_a": "A"} # 更新数据库逻辑 db_entry = get_entry(...) for name, value in update_data.items(): setattr(db_entry, name, value)
此方案完全利用Pydantic原生能力,无需额外代码,是最简洁通用的解决方案。
方案2:给字段设置default=Undefined
如果不想依赖exclude_unset,可以给模型字段设置default=Undefined,未传入的字段不会出现在模型的字典结果中:
from pydantic import BaseModel, Undefined, Field from typing import Optional class EntryToUpdateSchema(BaseModel): value_a: Optional[str] = Field(default=Undefined) value_b: Optional[str] = Field(default=Undefined) # 仅传入value_a时 entry_to_update = EntryToUpdateSchema(value_a="A") # entry_to_update.dict() 结果:{"value_a": "A"} # 显式传入value_b=None时 entry_to_update_with_none = EntryToUpdateSchema(value_b=None) # entry_to_update_with_none.dict() 结果:{"value_b": None}
之后直接遍历dict()的键值对更新即可,未传入的字段不会被处理。
方案3:自定义标记类(备选)
如果上述Pydantic方案不适用,可以自定义标记类区分“显式None”和“未设置”:
class ExplicitNone: pass # 接口层处理请求数据,将用户传入的None替换为标记类实例 def parse_request_data(raw_data): parsed_data = {} for key, val in raw_data.items(): parsed_data[key] = ExplicitNone() if val is None else val return EntryToUpdateSchema(**parsed_data) # 更新数据库逻辑 db_entry = get_entry(...) for name, value in entry_to_update.dict().items(): if isinstance(value, ExplicitNone): setattr(db_entry, name, None) elif value is not None: setattr(db_entry, name, value)
此方案需要额外的请求处理逻辑,仅在特殊场景下使用。
内容的提问来源于stack exchange,提问作者Palmos
相关产品推荐
相关产品推荐

