FastAPI返回的DateTime时区与数据库服务器响应不一致问题

问题根源 #

1. Pydantic序列化行为：默认情况下，Pydantic会将带时区的datetime对象转换为UTC时区输出（格式如YYYY-MM-DDTHH:MM:SSZ，对应GMT+0），不会保留原时区的偏移信息。
2. 静态默认时间：代码中SQLAlchemy列和Pydantic模型的creation_time默认值是在模块加载时一次性计算的固定值，而非每次请求时生成的当前时间，这会导致所有未指定该字段的请求都使用同一个时间。

分步解决 #

1. 让Pydantic保留时区偏移输出 #

修改Computer和ComputerIn模型，配置自定义JSON编码器，让序列化时输出带时区偏移的完整格式：

# 若使用Pydantic v2
class Computer(BaseModel):
    id: int
    computername: str
    computerip: str
    computerexternalip: str
    creation_time: datetime

    model_config = {
        "json_encoders": {
            datetime: lambda dt: dt.isoformat()
        }
    }

class ComputerIn(BaseModel):
    computername: str
    computerip: str
    computerexternalip: str
    creation_time: datetime = Field(default_factory=get_ny_time)

# 若使用Pydantic v1
class Computer(BaseModel):
    id: int
    computername: str
    computerip: str
    computerexternalip: str
    creation_time: datetime

    class Config:
        json_encoders = {
            datetime: lambda dt: dt.isoformat()
        }

2. 修复动态生成默认时间的问题 #

先定义一个生成纽约时区当前时间的可调用函数：

def get_ny_time():
    return datetime.utcnow().replace(tzinfo=pytz.UTC).astimezone(pytz.timezone("America/New_York"))

修改SQLAlchemy列的默认值

将静态时间替换为可调用函数，确保每次插入时生成新时间：

computers = sqlalchemy.Table(
    "computers32",
    metadata,
    sqlalchemy.Column("id", sqlalchemy.Integer, primary_key=True),
    sqlalchemy.Column("creation_time", sqlalchemy.DateTime(timezone=True), default=get_ny_time)
)

或者更推荐直接用数据库层面的时区函数生成时间（避免Python和数据库的时区差异）：

sqlalchemy.Column(
    "creation_time", 
    sqlalchemy.DateTime(timezone=True), 
    default=sqlalchemy.text("CURRENT_TIMESTAMP AT TIME ZONE 'America/New_York'")
)

修改Pydantic模型的默认值

使用default_factory替代直接赋值，确保每次请求生成新时间：

from pydantic import Field

class ComputerIn(BaseModel):
    computername: str
    computerip: str
    computerexternalip: str
    creation_time: datetime = Field(default_factory=get_ny_time)

3. 确保返回与数据库一致的时间（可选） #

当前代码返回的是请求传入或Pydantic生成的时间，若想完全匹配数据库存储的值，可以插入后查询数据库再返回：

@app.post("/computers/", response_model=Computer)
async def create_computer(computer: ComputerIn):
    query = computers.insert().values(creation_time=computer.creation_time)
    last_record_id = await database.execute(query)
    # 查询刚插入的记录
    select_query = computers.select().where(computers.c.id == last_record_id)
    db_computer = await database.fetch_one(select_query)
    return db_computer

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