如何使用Python对PostgreSQL查询按日期过滤且无需强转字段类型

你遇到的TypeError: Object of type 'datetime' is not JSON serializable报错并非来自数据库查询过滤环节：PostgreSQL会自动将你传入的字符串格式日期转为timestamp类型完成过滤，逻辑本身没有问题。报错触发在结果序列化阶段：未强转varchar时，查询返回的created_at、updated_at字段对应值为Python原生datetime类型，该类型没有默认的JSON序列化规则，后续对返回结果做JSON序列化操作时就会抛出错误。之前强转varchar是在数据库层面提前把时间转为字符串，避开了序列化问题。

以下3种常用方案均不需要在SQL中对时间字段做强转：

方案1：查询后手动转换datetime字段为字符串 #

优先推荐该方案，灵活度最高，可自定义时间输出格式。

1. 先修改SQL，移除字段强转逻辑：

SELECT
    id,
    created_at,
    updated_at,
    amount
FROM
    public.table_1
WHERE created_at >= %s and created_at <= %s;

2. 在Python代码中拿到查询结果后，手动转换时间字段为字符串：

def read_latest_data_from_pg(**kwargs):
    with open('dags/scripts/sql_scripts/pg_export_sql/sql_file.sql','r') as sqlfile:
            pg_export_data_query = sqlfile.read()
    max_1 = '2021-05-01'
    max_2 = '2021-05-28'
    pg_hook = PostgresHook(postgres_conn_id='pg_conn', delegate_to=None, use_legacy_sql=False)
    conn = pg_hook.get_conn()
    cursor = conn.cursor()
    cursor.execute(pg_export_data_query, (max_1, max_2))
    result = cursor.fetchall()
    
    # 新增时间字段转换逻辑，假设返回字段顺序为id、created_at、updated_at、amount
    processed_result = []
    for row in result:
        row_list = list(row)
        # 自定义时间输出格式，按需调整
        row_list[1] = row_list[1].strftime("%Y-%m-%d %H:%M:%S")
        row_list[2] = row_list[2].strftime("%Y-%m-%d %H:%M:%S")
        processed_result.append(tuple(row_list))
    
    print('result', processed_result)
    return processed_result, len(processed_result)

方案2：自定义JSON序列化器 #

如果你的场景是返回结果会直接传入json.dumps做序列化，可直接新增datetime类型的序列化规则，不需要修改查询和结果处理逻辑：

import json
from datetime import datetime

def datetime_serializer(obj):
    if isinstance(obj, datetime):
        # 自定义时间输出格式
        return obj.strftime("%Y-%m-%d %H:%M:%S")
    raise TypeError(f"Type {type(obj)} not serializable")

# 序列化时指定自定义规则即可
json_str = json.dumps(result, default=datetime_serializer)

方案3：配置psycopg2类型转换，直接返回字符串格式时间 #

可以通过配置PostgreSQL Python驱动的类型适配规则，让查询结果中的timestamp字段直接返回字符串，不需要额外处理：

import psycopg2
from psycopg2 import extensions

def read_latest_data_from_pg(**kwargs):
    with open('dags/scripts/sql_scripts/pg_export_sql/sql_file.sql','r') as sqlfile:
            pg_export_data_query = sqlfile.read()
    max_1 = '2021-05-01'
    max_2 = '2021-05-28'
    pg_hook = PostgresHook(postgres_conn_id='pg_conn', delegate_to=None, use_legacy_sql=False)
    conn = pg_hook.get_conn()
    
    # 新增类型转换注册逻辑
    def cast_timestamp(value, cur):
        return value if value is None else value
    TIMESTAMP_TYPE = psycopg2.extensions.new_type((1114,), "TIMESTAMP", cast_timestamp)
    psycopg2.extensions.register_type(TIMESTAMP_TYPE, conn)
    
    cursor = conn.cursor()
    cursor.execute(pg_export_data_query, (max_1, max_2))
    result = cursor.fetchall()
    print('result', result)
    return result, len(result)

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