使用 Psycopg3 插入带有 PostGIS 点的多行

Question

我正在使用 psycopg3 使用以下函数将多行插入到表中：

async def execute_values(self, command_with_placeholders, values) -> None:
    placeholders = sql.SQL(', ').join(sql.Placeholder() * len(values[0]))
    command = sql.SQL(command_with_placeholders).format(placeholders=placeholders)
    async with (await self.conn).cursor() as cursor:
        await cursor.executemany(command, values)

当我包含非空时，使用以下参数调用它会失败

location

：

insert_sessions_command = """
        INSERT INTO sessions (session_id, start_time, last_activity_time, end_time, ip,
            user_agent, country_code, region_code, location)
        VALUES ({placeholders});
    """
    await sql_db.execute_values(insert_sessions_command, [
        ('s1', datetime(2023, 1, 1), datetime(2023, 1, 2), None, '1.2.3.4', None,
         'NZ', 'NZ-CAN', 'ST_SetSRID(ST_MakePoint(172.6306, 43.5320), 4326)'),
    ]

使用 Postgres 控制台执行以下操作是有效的：

INSERT INTO sessions (session_id, start_time, last_activity_time, end_time, ip,
                user_agent, country_code, region_code, location)
            VALUES ('s1', '2023-01-01', '2023-01-02', null, '1.2.3.4', null,
             'NZ', 'NZ-CAN', ST_SetSRID(ST_MakePoint(172.6306, 43.5320), 4326));

所以我猜 psycopg 代码会失败，因为它用引号将点括起来，就像以下错误的 Postgres 命令：

INSERT INTO sessions (session_id, start_time, last_activity_time, end_time, ip,
                user_agent, country_code, region_code, location)
            VALUES ('s1', '2023-01-01', '2023-01-02', null, '1.2.3.4', null,
             'NZ', 'NZ-CAN', 'ST_SetSRID(ST_MakePoint(172.6306, 43.5320), 4326)');

在这两种情况下，错误都是

parse error - invalid geometry
HINT:  "ST" <-- parse error at position 2 within geometry
CONTEXT:  unnamed portal parameter $9 = '...'

如何使 Psycopg 代码不将点字符串用引号括起来来修复它？

Answer 1

作为简单的例子来展示我的意思：

import psycopg
from psycopg import sql

con = psycopg.connect('dbname=test host=localhost user=postgres')
cur = con.cursor()

sql_str = sql.SQL('select {round}({random}()::numeric, %s)')
cur.execute(sql_str.format(round=sql.Identifier('round'), random=sql.Identifier('random')), [3]).fetchone()
(Decimal('0.357'),)

cur.execute(sql_str.format(round=sql.Identifier('round'), random=sql.Identifier('random')), [2]).fetchone()
(Decimal('0.50'),)