FastAPI中POST请求的datetime序列化问题及解决方法咨询
问题描述
请求示例
curl -X 'POST' \ 'http://0.0.0.0:8000/api/v1/tasks/' \ -H 'accept: application/json' \ -H 'Content-Type: application/json' \ -d '{ "title": "string", "description": "string", "deadline": "2023-06-25T15:31:44.204Z" }'
错误日志
sqlalchemy.exc.DBAPIError: (sqlalchemy.dialects.postgresql.asyncpg.Error) <class 'asyncpg.exceptions.DataError'>: invalid input for query argument $3: '2023-06-25T15:31:44.204000+00:00' (expected a datetime.date or datetime.datetime instance, got 'str') [SQL: INSERT INTO task (title, description, deadline) VALUES ($1::VARCHAR, $2::VARCHAR, $3::TIMESTAMP WITHOUT TIME ZONE) RETURNING task.id] [parameters: ('string', 'string', '2023-06-25T15:31:44.204000+00:00')]
相关代码
数据库模型
class Task(Base): id: Mapped[int] = mapped_column(primary_key=True, index=True) title: Mapped[str] = mapped_column(String(40)) description: Mapped[str] deadline: Mapped[datetime] performers = relationship( "User", secondary=association_table, back_populates="tasks", )
Pydantic Schema
class TaskBase(BaseModel): title: str | None = None description: str | None = None deadline: datetime | None = None class TaskCreate(TaskBase): title: str description: str # performers: list[UserBase] | None = None class Config: orm_mode = True
CRUD代码
class CRUDTask(CRUDBase[Task, TaskCreate, TaskUpdate]): async def create(self, db: AsyncSession, obj_in: TaskCreate) -> Task: obj_in_data = jsonable_encoder(obj_in) # obj_in_data["deadline"] = datetime.strptime( # obj_in_data["deadline"], "%Y-%m-%dT%H:%M:%S.%f+00:00" # ) db_obj = self.model(**obj_in_data) db.add(db_obj) await db.commit() await db.refresh(db_obj) return db_obj
解决方案
问题核心是jsonable_encoder把Pydantic已解析好的datetime对象转成了ISO字符串,而SQLAlchemy需要的是datetime实例。正确做法是直接使用Pydantic模型自带方法获取字段数据,保留原始类型:
1. 修改CRUD的create方法
替换jsonable_encoder为Pydantic的model_dump()(v2版本)或dict()(v1版本):
class CRUDTask(CRUDBase[Task, TaskCreate, TaskUpdate]): async def create(self, db: AsyncSession, obj_in: TaskCreate) -> Task: # Pydantic v2用model_dump,v1用dict,exclude_unset=True可选,只传递已设置的字段 obj_in_data = obj_in.model_dump(exclude_unset=True) db_obj = self.model(**obj_in_data) db.add(db_obj) await db.commit() await db.refresh(db_obj) return db_obj
2. 版本适配说明
- 若使用Pydantic v1,将代码中的
model_dump替换为dict即可,参数保持一致; - 你的Pydantic Schema已正确将
deadline定义为datetime类型,FastAPI会自动把请求中的ISO字符串解析成datetime对象,无需额外处理。
为什么之前的方案不好?
jsonable_encoder的作用是把对象转成JSON可序列化格式(比如把datetime转成字符串),不适合直接传递给SQLAlchemy模型;- 手动解析字符串属于重复劳动,Pydantic已经完成了ISO字符串到datetime对象的转换工作。
内容的提问来源于stack exchange,提问作者yuwisasha
相关产品推荐
相关产品推荐

