现象描述
当 FastApi 访问数据库或其他遇到 datetima 时间场景时，若直接返回响应结果，时间会被自动格式为："2024-01-01T09:30:14",而不是 "2024-01-01 09:30:14"
本文包含2种处理方法，分别对应2种响应模型，解决办法，看这里 解决方法
第1种，直接响应 pydantic 模型
代码如下
SQLAlchemy 模型
"""
@ File : test_model.py
@ Author : yqbao
@ Version : V1.0.0
"""
import time
from sqlalchemy import Column, Integer, String, DateTime, Enum, SmallInteger
from db import Base
class TestTable(Base):
__tablename__ = 'test_table'
id = Column(Integer, primary_key=True, autoincrement=True, nullable=False)
name = Column(String(10), nullable=False, comment='名称')
createTime: str = Column(DateTime, nullable=False, default=time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()))
deleted = Column(Enum('0', '1'), nullable=False, server_default='0')
pydantic 模型
"""
@ File : test_schemas.py
@ Author : yqbao
@ Version : V1.0.0
"""
from datetime import datetime
from typing import Union
from pydantic import BaseModel, ConfigDict
class TestTable(BaseModel):
name: str
createTime: datetime
# from_attributes 将告诉 Pydantic 模型读取数据，即它不是一个 dict，而是一个 ORM 模型
# https://docs.pydantic.dev/2.0/usage/models/#arbitrary-class-instances
model_config = ConfigDict(
from_attributes=True
)
数据交互动作
"""
@ File : test_crud.py
@ Author : yqbao
@ Version : V1.0.0
"""
from sqlalchemy.orm import Session
from sqlalchemy.sql import and_
from test import test_model
class TestTableCRUD(object):
@staticmethod
def get_test_tables(db: Session, skip: int = 0, limit: int = 100):
"""项目列表"""
test_tables = db.query(test_model.TestTable).filter(
test_model.TestTable.deleted == '0').offset(skip).limit(limit).all()
return test_tables
@staticmethod
def get_test_table(db: Session, test_table_id: int):
"""ID查询"""
test_table = db.query(test_model.TestTable).filter(
and_(test_model.TestTable.id == test_table_id, test_model.TestTable.deleted == '0')).first()
return test_table
添加路由
"""
@ File : test_table.py
@ Author : yqbao
@ Version : V1.0.0
"""
from fastapi import APIRouter, Depends
from sqlalchemy.orm import Session
from db import get_db
from test import test_schemas, test_crud
router = APIRouter(
prefix="/test-table",
tags=["test-table"],
responses={404: {"message": "Not Found"}}
)
crud = test_crud.TestTableCRUD()
# 注意此处的相应模型为：response_model=list[test_schemas.TestTable]
# 直接响应 pydantic 模型
@router.get("/list", response_model=list[test_schemas.TestTable], tags=["test-table"])
async def read_test_tables(skip: int = 0, limit: int = 20, db: Session = Depends(get_db)):
db_test_tables = crud.get_test_tables(db, skip=skip, limit=limit)
return db_test_tables
@router.get("/{test_table_id}", response_model=test_schemas.TestTable, tags=["test-table"])
async def read_test_table(test_table_id: int, db: Session = Depends(get_db)):
db_test_table = crud.get_test_table(db, test_table_id)
return db_test_table
其余代码省略，然后启动服务并访问/docs文档，得到如下2个路由
访问：/test-table/list?skip=0&limit=20 得到的响应体如下:
第2种，响应自定义公共模型
代码如下（仅展示与第1种不相同的代码）
pydantic 模型
"""
@ File : test_schemas.py
@ Author : yqbao
@ Version : V1.0.0
@ Description : 创建初始 Pydantic模型,用于关联ORM
"""
from datetime import datetime
from typing import Union
from pydantic import BaseModel, ConfigDict
# 添加公共响应模型
class ResponseBase(BaseModel):
code: int
message: str
data: Union[list, dict, str] = []
自定义响应
"""
@ File : response.py
@ Author : yqbao
@ Version : V1.0.0
@ Description :
"""
from typing import Union
from fastapi.responses import JSONResponse
from fastapi import status
# 200
def response200(data: Union[list, dict, str], message: str = 'Success') -> JSONResponse:
return JSONResponse(
status_code=status.HTTP_200_OK,
content={
'code': 0,
'message': message,
'data': data
}
)
# 500
def response500(data: str = '', message: str = 'Internal Server Error') -> JSONResponse:
return JSONResponse(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR_CODE,
content={
'code': 1,
'message': message,
'data': data
}
)
路由
"""
@ File : test_table.py
@ Author : yqbao
@ Version : V1.0.0
@ Description :
"""
from fastapi import APIRouter, Depends
from fastapi.encoders import jsonable_encoder
from sqlalchemy.orm import Session
from db import get_db
from extension import response
from test import test_schemas, test_crud
router = APIRouter(
prefix="/test-table",
tags=["test-table"],
responses={404: {"message": "Not Found"}}
)
crud = test_crud.TestTableCRUD()
# 响应模型切换为公共响应：response_model=test_schemas.ResponseBase
@router.get("/list", response_model=test_schemas.ResponseBase, tags=["test-table"])
async def read_test_tables(skip: int = 0, limit: int = 20, db: Session = Depends(get_db)):
db_test_tables = crud.get_test_tables(db, skip=skip, limit=limit)
# 使用 FastApi 提供的json编码器，并交由自定义响应来返回
db_test_tables = jsonable_encoder(db_test_tables)
return response.response200(data=db_test_tables)
@router.get("/{test_table_id}", response_model=test_schemas.ResponseBase, tags=["test-table"])
async def read_test_table(test_table_id: int, db: Session = Depends(get_db)):
db_test_table = crud.get_test_table(db, test_table_id)
db_test_table = jsonable_encoder(db_test_table)
return response.response200(data=db_test_table)
访问：/test-table/list?skip=0&limit=20 得到的响应体如下:
定位问题
时间会被自动格式为："2024-01-01T09:30:14"，这是由 Pydantic 所决定的，详见：Datetimes
解决办法
第1种，直接响应 pydantic 模型
修改代码如下
pydantic 模型
"""
@ File : test_schemas.py
@ Author : yqbao
@ Version : V1.0.0
@ Description : 创建初始 Pydantic模型,用于关联ORM
"""
from datetime import datetime
from typing import Union
from pydantic import BaseModel, ConfigDict
class TestTable(BaseModel):
name: str
createTime: datetime
# from_attributes 将告诉 Pydantic 模型读取数据，即它不是一个 dict，而是一个 ORM 模型
# https://docs.pydantic.dev/2.0/usage/models/#arbitrary-class-instances
# 添加自定义 json编码器
model_config = ConfigDict(
from_attributes=True,
json_encoders={datetime: lambda dt: dt.strftime("%Y-%m-%d %H:%M:%S")}
)
重新访问：/test-table/list?skip=0&limit=20 得到的响应体如下:
第2种，响应自定义公共模型
修改代码如下：
路由
"""
@ File : test_table.py
@ Author : yqbao
@ Version : V1.0.0
@ Description :
"""
from datetime import datetime
from fastapi import APIRouter, Depends
from fastapi.encoders import jsonable_encoder
from sqlalchemy.orm import Session
from db import get_db
from extension import response
from test import test_schemas, test_crud
router = APIRouter(
prefix="/test-table",
tags=["test-table"],
responses={404: {"message": "Not Found"}}
)
crud = test_crud.TestTableCRUD()
# 添加自定义编码器
custom_encoder = {datetime: lambda dt: dt.strftime("%Y-%m-%d %H:%M:%S")}
@router.get("/list", response_model=test_schemas.ResponseBase, tags=["test-table"])
async def read_test_tables(skip: int = 0, limit: int = 20, db: Session = Depends(get_db)):
db_test_tables = crud.get_test_tables(db, skip=skip, limit=limit)
db_test_tables = jsonable_encoder(db_test_tables, custom_encoder=custom_encoder)
return response.response200(data=db_test_tables)
@router.get("/{test_table_id}", response_model=test_schemas.ResponseBase, tags=["test-table"])
async def read_test_table(test_table_id: int, db: Session = Depends(get_db)):
db_test_table = crud.get_test_table(db, test_table_id)
db_test_table = jsonable_encoder(db_test_table, custom_encoder=custom_encoder)
return response.response200(data=db_test_table)
重新访问：/test-table/list?skip=0&limit=20 得到的响应体如下:
