如何使用pg8000向PostgreSQL存储过程传递含NULL的数组？

问题背景 #

在Python 3.8环境下，使用pg8000 1.21.3调用PostgreSQL 11的存储过程，该过程接收自定义数据类型数组作为参数。直接通过SQL调用时可正常处理NULL值，但通过Python代码手动替换None为空字符串的方式会触发布尔、小数类型的解析错误。

可用SQL架构及存储过程 #

DROP PROCEDURE IF EXISTS myproc;
DROP TABLE IF EXISTS mytable;
DROP TYPE mytype;

CREATE TYPE mytype AS (
    an_int     INT,
    a_bool     BOOL,
    a_decimal  DECIMAL,
    a_string   TEXT
);

CREATE TABLE mytable(
    id         SERIAL PRIMARY KEY,
    an_int     INT,
    a_bool     BOOL,
    a_decimal  DECIMAL,
    a_string   TEXT
);

CREATE PROCEDURE myproc(
    IN myarray mytype[]
)
AS $$
BEGIN
        INSERT INTO mytable(
            an_int,
            a_bool,
            a_decimal,
            a_string
        ) SELECT
            an_int,
            a_bool,
            a_decimal,
            a_string
        FROM
            unnest(myarray);
    END;
$$
LANGUAGE 'plpgsql';

-- 直接SQL调用可正常处理NULL
CALL myproc(
    array[
            (1, 1, 2.1, 'foo'),
            (NULL, NULL, NULL, NULL)
        ]::mytype[]
);

SELECT * FROM mytable;

错误的Python实现 #

以下代码通过替换None为空字符串传递参数，会触发布尔、小数类型的解析错误：

import pg8000

arr = [(1, True, 3.2, 'foo'),
       (None, True, 3.2, 'foo'),
       (1, True, 3.2, None),
       (1, False, 3.2, 'foo'),
       (1, None, 3.2, 'foo'),  # 布尔类型NULL报错
        (1, False, None, 'foo'), # 小数类型NULL报错
       ]

sql = """CALL myproc(%s::mytype[]);"""
stm = [str(item) for item in arr]
stm = [item.replace('None', '') for item in stm]

with pg8000.connect(host="localhost", database="mydb", user="myuser", password="mypass") as con:
    with con.cursor() as cursor:
        cursor.execute(
            sql,
            args=(stm,)
        )
    con.commit()

报错信息 #

pg8000.dbapi.ProgrammingError: {'S': 'ERROR', 'V': 'ERROR', 'C': '22P02', 'M': 'invalid input syntax for type boolean: " "', 'F': 'bool.c', 'L': '154', 'R': 'boolin'}

解决方案 #

核心是避免手动拼接字符串，利用pg8000对PostgreSQL复合类型和数组的原生支持，让驱动自动将Python的None映射为PostgreSQL的NULL。

方法1：使用字典构造自定义类型 #

import pg8000
from pg8000.types import Array

# 构造符合mytype结构的字典列表，直接保留None
arr = [
    {"an_int": 1, "a_bool": True, "a_decimal": 3.2, "a_string": "foo"},
    {"an_int": None, "a_bool": True, "a_decimal": 3.2, "a_string": "foo"},
    {"an_int": 1, "a_bool": True, "a_decimal": 3.2, "a_string": None},
    {"an_int": 1, "a_bool": False, "a_decimal": 3.2, "a_string": "foo"},
    {"an_int": 1, "a_bool": None, "a_decimal": 3.2, "a_string": "foo"},
    {"an_int": 1, "a_bool": False, "a_decimal": None, "a_string": "foo"},
]

# 无需手动转换类型，pg8000会自动处理
sql = "CALL myproc(%s);"

with pg8000.connect(host="localhost", database="mydb", user="myuser", password="mypass") as con:
    with con.cursor() as cursor:
        cursor.execute(sql, args=(Array(arr),))
    con.commit()

方法2：使用命名元组构造自定义类型 #

命名元组更贴近SQL复合类型的结构，可读性更强：

import pg8000
from collections import namedtuple
from pg8000.types import Array

# 定义与mytype结构匹配的命名元组
MyType = namedtuple("MyType", ["an_int", "a_bool", "a_decimal", "a_string"])

arr = [
    MyType(1, True, 3.2, "foo"),
    MyType(None, True, 3.2, "foo"),
    MyType(1, True, 3.2, None),
    MyType(1, False, 3.2, "foo"),
    MyType(1, None, 3.2, "foo"),
    MyType(1, False, None, "foo"),
]

sql = "CALL myproc(%s);"

with pg8000.connect(host="localhost", database="mydb", user="myuser", password="mypass") as con:
    with con.cursor() as cursor:
        cursor.execute(sql, args=(Array(arr),))
    con.commit()

原理说明 #

之前的错误在于手动将None替换为空字符串，导致PostgreSQL解析时把空字符串当作布尔/小数类型的输入值，而这些类型不接受空字符串作为合法输入。通过pg8000的Array类型包装结构化数据（字典或命名元组），驱动会自动完成：

1. 将Python的None转换为PostgreSQL的NULL
2. 将结构化数据映射为PostgreSQL的mytype复合类型
3. 将列表转换为mytype[]数组类型

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