使用SQLAlchemy与psycopg向PostgreSQL写入DataFrame遇类型适配错误

从Kobo Toolbox拉取数据生成DataFrame后，使用SQLAlchemy将数据写入PostgreSQL数据库时触发该错误，其他拉取数据的步骤正常，代码如下：

import psycopg2
import pandas as pd
from sqlalchemy import create_engine
import sqlalchemy
from sqlalchemy.types import String
import pykobo


# establish connections
conn_string = 'postgresql://postgres:openppp@localhost:5432/wpp_appeals'

db = create_engine(conn_string)
conn = db.connect()


URL_KOBO_API = "https://kf.kobotoolbox.org/api/v2"
MYTOKEN = "1e83cdede6c48a19ef210e417d6ca61b076db175"
km = pykobo.Manager(url_api=URL_KOBO_API, token=MYTOKEN)

uid = 'aQCxZ9BVrHe33pLMdtx3gV'
my_form = km.get_form(uid)

    
my_form.fetch_data()  
df1 = my_form.data


df1.to_sql('appeals', con=conn, if_exists='append', index=False, dtype={
    'deviceid':sqlalchemy.String(),
    'TimeStartRecorded':sqlalchemy.VARCHAR(),
    'TimeEndRecorded':sqlalchemy.VARCHAR(),
    'today':sqlalchemy.VARCHAR(),
    'Appeals_Feedback':sqlalchemy.VARCHAR(),
    'Location':sqlalchemy.VARCHAR(),
    'gps_location':sqlalchemy.VARCHAR(),
    '_gps_location_latitude':sqlalchemy.VARCHAR(),
    '_gps_location_longitude':sqlalchemy.VARCHAR(),
    '_gps_location_altitude':sqlalchemy.VARCHAR(),
    '_gps_location_precision':sqlalchemy.VARCHAR(),
    'householdID':sqlalchemy.VARCHAR(),
    'hhNumber':sqlalchemy.VARCHAR(),
    'confirm1':sqlalchemy.VARCHAR(),
    'focalPointName':sqlalchemy.VARCHAR(),
    'validz':sqlalchemy.VARCHAR(),
    '_id':sqlalchemy.VARCHAR(),
    '_uuid':sqlalchemy.VARCHAR(),
    '_submission_time':sqlalchemy.VARCHAR(),
    '_validation_status':sqlalchemy.VARCHAR(),
    '_notes':sqlalchemy.VARCHAR(),
    '_status':sqlalchemy.VARCHAR(),
    '_submitted_by':sqlalchemy.VARCHAR(),
    '_tags':sqlalchemy.VARCHAR(),
    '_index':sqlalchemy.VARCHAR()})

conn = psycopg2.connect(conn_string
                        )

conn.autocommit = True
                     


#conn.commit()
conn.close()

原因分析 #

Kobo Toolbox返回的DataFrame中存在字典类型的列，尽管你在dtype参数中指定了部分列的类型，但未覆盖所有含dict数据的列，psycopg2无法直接将Python dict类型转换为PostgreSQL支持的原生数据类型，因此触发该错误。

解决方案 #

1. 定位字典类型列 #

先运行以下代码找出DataFrame中包含dict数据的列：

for col in df1.columns:
    unique_types = df1[col].apply(type).unique()
    print(f"列名: {col}, 数据类型: {unique_types}")

2. 处理字典类型数据 #

根据业务需求选择以下方式之一处理：

• 序列化为JSON字符串：如果需要保留嵌套结构，将dict转为JSON字符串（可存入PostgreSQL的TEXT或JSONB类型）：

  import json
  # 替换为实际的字典列名
  df1['Location'] = df1['Location'].apply(lambda x: json.dumps(x) if isinstance(x, dict) else x)
  

• 直接删除列：如果该列数据无业务价值，直接删除：

  df1 = df1.drop(columns=['包含dict的列名'])
  

3. 更新to_sql的dtype配置 #

确保所有列都被正确指定类型，若选择存储JSON，可使用SQLAlchemy的JSON类型：

from sqlalchemy.types import JSON, String

df1.to_sql('appeals', con=db, if_exists='append', index=False, dtype={
    # 原有列配置...
    'Location': JSON,  # 若存储为JSON类型
    # 或 'Location': String()  # 若存储为JSON字符串
})

4. 优化数据库连接代码 #

代码中同时创建了SQLAlchemy连接和psycopg2连接，存在冗余，只需保留SQLAlchemy引擎即可：

# 移除多余的psycopg2连接代码
# conn = psycopg2.connect(conn_string)
# conn.autocommit = True

# 使用SQLAlchemy引擎直接入库
df1.to_sql('appeals', con=db, if_exists='append', index=False, dtype={
    # 你的类型配置
})

conn.close()

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