Python3.9用psycopg2操作PostgreSQL13+PostGIS出现SQL语法错误

直接语法错误原因 #

PostgreSQL 标准 UPDATE 语句不需要加 TABLE 关键字，错误行里的UPDATE TABLE写法不符合语法规范，正确格式为 UPDATE 表名 SET 字段=值 WHERE 条件。

其他潜在错误 #

• 字符串拼接传参风险：直接拼接Geometry类型的文本内容，如果内容中包含单引号等特殊字符，会直接触发SQL语法错误，同时存在SQL注入漏洞。
• 游标结果集单次遍历问题：pcur.fetchall()执行一次后游标会移动到结果集末尾，第二次及之后的循环遍历时会返回空列表，内层统计逻辑无法正常执行。
• 分表创建逻辑错误：parallelJoin函数中创建rf分表时，仍然查询的是点表pointsTable，应该替换为面表rectsTable。
• 线程安全问题：psycopg2的数据库连接不是线程安全的，多线程共享同一个连接会导致不可预期的交互异常。

• 打印实际执行的SQL：在报错的execute执行前，先打印拼接好的完整SQL语句，复制到psql或pgAdmin中直接运行，PostgreSQL会返回精准的错误位置和原因。
• 捕获数据库异常：用try-except块包裹SQL执行逻辑，捕获psycopg2.Error类型的异常，直接打印异常对象即可获取数据库返回的完整报错详情。
• 使用参数化查询：所有传入SQL的变量都通过psycopg2的参数占位符%s传递，参数放在execute方法的第二个参数元组中，框架会自动处理类型转换和特殊字符转义。
• 简化场景测试：先注释掉循环、多线程逻辑，用固定的单条参数测试SQL执行，确认逻辑正确后再逐步恢复上层逻辑。

import psycopg2
import threading

def corefunc(rf, openConnection):
    pcur = openConnection.cursor(name="pcur" + rf)
    rcur = openConnection.cursor(name="rcur" + rf)
    acur = openConnection.cursor()
    rcur.execute(f"SELECT geom FROM {rf}")
    # 提前加载r的所有geom避免重复fetch
    r_geoms = rcur.fetchall()

    for number in range (1, 5):
        acur.execute(f"DROP TABLE IF EXISTS pf{rf}")
        acur.execute(f"CREATE TABLE pf{rf} (index integer, sums integer)")
        pcur.execute(f"SELECT geom FROM pf{str(number)}")
        # 提前加载当前分块的所有点geom避免重复fetch
        p_geoms = pcur.fetchall()
        row = 1
        for each in r_geoms:
            if number == 1:
                acur.execute(f"INSERT INTO pf{rf} (index, sums) VALUES (%s, 0)", (row,))
            for eachone in p_geoms:
                # 修正UPDATE语法，使用参数化查询
                acur.execute(
                    f"UPDATE pf{rf} SET sums = sums + ST_Contains(%s, %s)::int WHERE index = %s",
                    (each[0], eachone[0], row)
                )
            row = row + 1
    openConnection.commit()

def parallelJoin (pointsTable, rectsTable, outputTable, outputPath, db_config):
    conn = psycopg2.connect(**db_config)
    cursor = conn.cursor()

    cursor.execute(f"SELECT COUNT(*) FROM {pointsTable}")
    size_data = (cursor.fetchall())[0][0]
    for number in range(1, 5):
        cursor.execute(f"DROP TABLE IF EXISTS pf{str(number)}")    
        cursor.execute(f"CREATE TABLE pf{str(number)} AS SELECT * FROM {pointsTable} LIMIT %s OFFSET %s",
                        (size_data/4, ((number-1)*size_data)/4))
    
    cursor.execute(f"SELECT COUNT(*) FROM {rectsTable}")
    size_rects = (cursor.fetchall())[0][0]
    for number in range(1, 5):
        cursor.execute(f"DROP TABLE IF EXISTS rf{str(number)}")
        # 修正为查询面表rectsTable
        cursor.execute(f"CREATE TABLE rf{str(number)} AS SELECT * FROM {rectsTable} LIMIT %s OFFSET %s",
                        (size_rects/4, ((number - 1) * size_rects)/4))
    conn.commit()
    threads = dict()
    for number in range(0, 4):
        # 每个线程单独创建连接保证线程安全
        thread_conn = psycopg2.connect(**db_config)
        threads[number] = threading.Thread(target=corefunc, args=(f"rf{str(number + 1)}", thread_conn))
        threads[number].start()
    
    for t in threads.values():
        t.join()
    
    # 后续逻辑

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