psycopg2 / SQLAlchemy：使用自定义类型数组参数执行函数

Question

我有一个自定义的postgres类型，看起来像这样：

CREATE TYPE "Sensor".sensor_telemetry AS
(
    sensorid character varying(50),
    measurement character varying(20),
    val numeric(7,3),
    ts character varying(20)
);

我正在尝试执行将这种类型的数组作为参数的postgres函数的调用。

我正在使用SQLAlchemy调用此函数，如下所示：

result = db.session.execute("""select "Sensor"."PersistTelemetryBatch"(:batch)""", batch)

batch看起来像：

{
    "batch" : [
        {
            "sensorID" : "phSensorA.haoshiAnalogPh",
            "measurement" : "ph",
            "value": 8.7,
            "timestamp": "2019-12-06 18:32:36"
        },
        {
            "sensorID" : "phSensorA.haoshiAnalogPh",
            "measurement" : "ph",
            "value": 8.8,
            "timestamp": "2019-12-06 18:39:36"
        }
    ]
}

运行此执行程序时，遇到此错误：

sqlalchemy.exc.ProgrammingError: (psycopg2.ProgrammingError) can't adapt type 'dict'

我猜测psycopg2抱怨自定义类型数组条目是一个字典，因为我可以将字典作为参数提供给其他pg函数执行（但是像这种情况下，这些字典不包含在数组中）。我对此是否正确？

我如何正确地将这些对象的数组传递给我的pg函数？

Answer 1

传递数据的一种直接方法是使用Python传递到convert the list of dicts to a list of tuples，并让psycopg2处理将其调整为适合的SQL构造：

from operator import itemgetter

ig = itemgetter("sensorID", "measurement", "value", "timestamp")
batch = {"batch": list(map(ig, batch["batch"]))}
query = """
        SELECT "Sensor"."PersistTelemetryBatch"(
            CAST(:batch AS "Sensor".sensor_telemetry[]))
        """
result = db.session.execute(query, batch)

[当您的数据是dict的列表时，另一个有趣的选择是使用json_populate_record()或json_populate_recordset()，但对于那些您必须固定键以使其匹配：

import json

batch = [{"sensorid": r["sensorID"], 
          "measurement": r["measurement"],
          "val": r["value"],
          "ts": r["timestamp"]}
         for r in batch["batch"]]
batch = {"batch": json.dumps(batch)}

query = """
        SELECT "Sensor"."PersistTelemetryBatch"(
            (SELECT ARRAY(SELECT rec
                          FROM json_populate_recordset(
                                   null::"Sensor".sensor_telemetry,
                                   :batch) AS rec)))
        """
result = db.session.execute(query, batch)