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

【问题标题】：psycopg2/SQLAlchemy: execute function with custom type array parameterpsycopg2/SQLAlchemy：使用自定义类型数组参数执行函数
【发布时间】：2019-12-09 00:17:45
【问题描述】：

我有一个如下所示的自定义 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 抱怨自定义类型数组条目是 dict，因为我可以将字典作为参数提供给其他 pg 函数执行（但这些字典不包含在这种情况下的数组中）。我说的对吗？

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

【问题讨论】：

将您的字典转换为元组：stackoverflow.com/questions/59031648/…
使用链接答案中的代码（psycopg/literal 响应），元组列表看起来像 [('value', 'measurement', 'timestamp', 'sensorID'), ('值'，'测量'，'时间戳'，'sensorID'）]。我应该在这里提供这些键的值吗？此外，每个元组中值的顺序是否需要与我创建的 pgsql 类型相同？ @IljaEverilä
元组应该包含值，并且它们应该反映您的用户定义类型的字段的顺序。想一想，这可能是 Python 的 operator.itemgetter() 的一个很好的用例，如果给定多个键，iirc 会返回值的元组。

标签： python postgresql sqlalchemy psycopg2

【解决方案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"(ARRAY(
            SELECT json_populate_recordset(
                       NULL::"Sensor".sensor_telemetry,
                       :batch)))
        """
result = db.session.execute(query, batch)

【讨论】：