Python环境下PostgreSQL查询返回datetime无法JSON序列化方案问询

问题核心说明 #

json.dumps 本身的作用就是将Python对象序列化为JSON字符串，所以你之前传入自定义Encoder或default=str得到字符串是符合预期的。你的核心需求是先将查询返回的RealDict中的datetime类型字段转为字符串，得到可直接序列化的普通Python字典，无需手动逐个处理字段。

最优解决方案（从数据源层面解决，推荐） #

直接在psycopg2连接层注册时间类型转换器，查询返回的时间字段自动转为字符串，后续无需任何额外处理：

import psycopg2
from datetime import datetime, date

# 在DatabaseConnector的connect方法中添加以下逻辑
def connect(self):
    # 原有连接逻辑不变，新增以下类型注册代码
    def cast_time_type(value, cur):
        if value is None:
            return None
        return str(value)
    
    # 注册timestamp、timestamptz、date类型的转换规则
    TIME_TYPE_OIDS = (1082, 1114, 1184) # 分别对应date、timestamp、timestamptz的PG内置OID
    time_cast = psycopg2.extensions.new_type(TIME_TYPE_OIDS, "TIME_TYPE", cast_time_type)
    psycopg2.extensions.register_type(time_cast)

配置完成后，fetchone()返回的结果中所有时间字段已经是字符串类型，直接调用json.dumps(fetch)即可正常运行，表结构新增时间字段也无需修改代码。

通用转换函数方案（不改连接层的选择） #

如果不想修改数据库连接配置，可以写一个递归转换函数，自动处理任意嵌套结构中的时间字段：

from datetime import datetime, date

def convert_datetime_to_str(obj):
    if isinstance(obj, dict):
        return {k: convert_datetime_to_str(v) for k, v in obj.items()}
    elif isinstance(obj, list):
        return [convert_datetime_to_str(item) for item in obj]
    elif isinstance(obj, (datetime, date)):
        return str(obj)
    # 可扩展其他需要转换的类型规则
    return obj

使用时只需要对查询结果套一层转换即可：

fetch = user.fetchone()
converted_fetch = convert_datetime_to_str(fetch)
# converted_fetch就是所有时间字段为字符串的普通字典，可直接序列化或使用
print(json.dumps(converted_fetch))

该方案支持单条、多条查询结果，嵌套结构也能正常处理，扩展性强。

项目级方案（适合Web服务场景） #

如果是正式项目，更推荐用Pydantic定义数据模型，自动处理类型转换同时支持字段校验：

from pydantic import BaseModel
from datetime import datetime

class UserModel(BaseModel):
    uname: str
    email: str
    fname: str
    lname: str
    create_date: datetime

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

# 直接将查询结果传入模型
user = UserModel(**fetch)
# 转字典（时间字段已自动转为字符串）
user_dict = user.dict()
# 直接转JSON字符串
user_json = user.json()

额外注意 #

你当前的插入SQL写法存在问题：'NOW()'加了单引号会被识别为普通字符串而非PG内置函数，应该去掉单引号，同时建议不要用f字符串拼接SQL，改用参数化查询避免SQL注入风险：

# 错误写法
db.execute_query(f"INSERT INTO user (uname, email, fname, lname, passwordhash, create_date) VALUES ('usera', 'usera@gmail.com', 'Jesse', 'Pinkman', 'myhash', 'NOW()');")
# 正确参数化写法
self._cur.execute(
    "INSERT INTO user (uname, email, fname, lname, passwordhash, create_date) VALUES (%s, %s, %s, %s, %s, NOW())",
    ('usera', 'usera@gmail.com', 'Jesse', 'Pinkman', 'myhash')
)

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