将空数组[]写入PostgreSQL text[]列时的SQLAlchemy报错问题

• 使用Python结合SQLAlchemy向PostgreSQL的text[]类型列（名为reply_to）写入数据时，空数组[]会触发sqlalchemy.exc.DataError错误，提示：

  (psycopg2.errors.InvalidTextRepresentation) malformed array literal: "[]"

• 非空数组可正常插入，尝试过以下方法均未解决：
  1. data["reply_to"] = data['reply_to'].replace('[]', '{}')
  2. data["reply_to"] = data['reply_to'].replace('[]', None)
  3. 在to_sql中指定dtype={'reply_to': sqlalchemy.types.JSON}

相关代码：

def write_data(table, data):
    conn_string = 'postgresql://creds/cryptodb'
    params = config()
    db = create_engine(conn_string)
    conn = db.connect()
    conn1 = psycopg2.connect(**params)
    if table == 'price_action_btc':
        data.reset_index(level=0, inplace=True)
        data.rename(columns={'Datetime': 'datetime', 'Open': 'open', 'Close': 'close', 'High': 'high', 'Low': 'low',
                             'Adj Close': 'adj_close', 'Volume': 'volume'}, inplace=True)
    else:
        data.drop('search', axis=1, inplace=True)
        data["conversation_id"] = pd.to_numeric(data["conversation_id"])
        data["created_at"] = pd.to_numeric(data["created_at"])
        data["date"] = pd.to_datetime(data["date"])
        data["retweet_date"] = pd.to_datetime(data["retweet_date"])
        data["user_id"] = pd.to_numeric(data["user_id"])
        data["day"] = pd.to_numeric(data["day"])
        data["hour"] = pd.to_numeric(data["hour"])
        data["video"] = pd.to_numeric(data["video"])
        data["nlikes"] = pd.to_numeric(data["nlikes"])
        data["nreplies"] = pd.to_numeric(data["nreplies"])
        data["nretweets"] = pd.to_numeric(data["nretweets"])
        data["retweet_id"] = pd.to_numeric(data["retweet_id"])
        data["user_rt_id"] = pd.to_numeric(data["user_rt_id"])
        # data["reply_to"] = data['reply_to'].replace('[]', '{}')
    for i in range(len(data)):
        try:
            data.iloc[i:i + 1].to_sql(name=table, if_exists='append', con=db, index=False, dtype={'reply_to': sqlalchemy.types.JSON})
        except exc.IntegrityError:
            pass
    conn1.commit()
    conn1.close()

DataFrame中reply_to列示例数据：

data['reply_to']
0                                                      []
1                                                      []
2       [{'screen_name': 'insiliconot', 'name': '핀핟핤핚핝...
3                                                      []
4                                                      []
                              ...                        
9995                                                   []

方法1：将字符串格式的数组转换为Python列表 #

问题核心是DataFrame中的[]是字符串类型，而非Python原生列表，PostgreSQL无法识别字符串"[]"为合法数组。使用ast.literal_eval将字符串安全转换为列表：

import ast

# 处理reply_to列：字符串转列表，空字符串[]转为空列表
data["reply_to"] = data["reply_to"].apply(
    lambda x: ast.literal_eval(x) if isinstance(x, str) and x.strip() else []
)

方法2：指定正确的PostgreSQL数组类型 #

在to_sql中使用SQLAlchemy针对PostgreSQL的ARRAY类型，而非JSON类型，匹配数据库的text[]列：

from sqlalchemy.dialects.postgresql import ARRAY
from sqlalchemy import Text

# 修改to_sql的dtype参数
data.to_sql(
    name=table,
    if_exists='append',
    con=db,
    index=False,
    dtype={'reply_to': ARRAY(Text)}
)

方法3：替换循环单条插入为批量插入 #

原代码循环逐条插入效率极低，且容易引发异常。结合前两种方法，改为批量插入：

import ast
from sqlalchemy.dialects.postgresql import ARRAY
from sqlalchemy import Text

def write_data(table, data):
    conn_string = 'postgresql://creds/cryptodb'
    params = config()
    db = create_engine(conn_string)
    
    if table == 'price_action_btc':
        data.reset_index(level=0, inplace=True)
        data.rename(columns={'Datetime': 'datetime', 'Open': 'open', 'Close': 'close', 'High': 'high', 'Low': 'low',
                             'Adj Close': 'adj_close', 'Volume': 'volume'}, inplace=True)
    else:
        data.drop('search', axis=1, inplace=True)
        # 处理各数值/日期列
        data["conversation_id"] = pd.to_numeric(data["conversation_id"])
        data["created_at"] = pd.to_numeric(data["created_at"])
        data["date"] = pd.to_datetime(data["date"])
        data["retweet_date"] = pd.to_datetime(data["retweet_date"])
        data["user_id"] = pd.to_numeric(data["user_id"])
        data["day"] = pd.to_numeric(data["day"])
        data["hour"] = pd.to_numeric(data["hour"])
        data["video"] = pd.to_numeric(data["video"])
        data["nlikes"] = pd.to_numeric(data["nlikes"])
        data["nreplies"] = pd.to_numeric(data["nreplies"])
        data["nretweets"] = pd.to_numeric(data["nretweets"])
        data["retweet_id"] = pd.to_numeric(data["retweet_id"])
        data["user_rt_id"] = pd.to_numeric(data["user_rt_id"])
        
        # 处理reply_to列：字符串转列表
        data["reply_to"] = data["reply_to"].apply(
            lambda x: ast.literal_eval(x) if isinstance(x, str) and x.strip() else []
        )
    
    # 批量插入
    try:
        data.to_sql(
            name=table,
            if_exists='append',
            con=db,
            index=False,
            dtype={'reply_to': ARRAY(Text)}
        )
    except exc.IntegrityError:
        pass
    
    # 原psycopg2连接未实际使用，可移除或调整
    # conn1 = psycopg2.connect(**params)
    # conn1.commit()
    # conn1.close()

说明：原代码中创建的conn1（psycopg2连接）未参与数据插入操作，可直接移除，避免资源浪费。

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