Python金仓数据库操作之SQL执行,批量操作与扩展功能的完全指南

一、SQL语句执行与结果集处理

在Python中操作金仓数据库，首先需要通过 ksycopg2.connect() 建立连接，并使用 cursor() 创建游标。执行SQL语句主要依赖 execute() 方法，而获取结果则根据数据量大小选择不同的策略。

1. 查询结果获取策略

• fetchone()：逐条获取结果，适合处理海量数据，能有效控制内存占用。
• fetchmany(size)：分批获取，通过指定批次大小平衡读取速度与内存消耗。
• fetchall()：一次性获取所有结果，仅建议用于数据量较小的场景。若结果集达数万条以上，应避免使用此方法以防内存溢出。

import ksycopg2

conn = ksycopg2.connect(
    database='TEST',
    user='SYSTEM',
    password='123456',
    host='127.0.0.1',
    port='54321'
)
cur = conn.cursor()

# 示例：分批获取数据
cur.execute("SELECT id, name FROM test_user")
while True:
    rows = cur.fetchmany(100)  # 每次获取100条
    if not rows:
        break
    for row in rows:
        process_row(row)

2. 元数据与类型信息

执行查询后，可通过 cursor.description 获取列的详细信息，包括列名、类型码（Kingbase内部OID）及长度。这对于动态处理结果集非常有用。

cur.execute("SELECT id, name, created_at FROM test_user")
for col in cur.description:
    print(f"列名: {col.name}, 类型码: {col.type_code}, 长度: {col.internal_size}")

3. 非查询语句与事务提交

对于 INSERT、UPDATE 和 DELETE 操作，执行后必须显式调用 conn.commit() 以提交事务，否则更改不会持久化。

cur = conn.cursor()
cur.execute("INSERT INTO test_user (id, name) VALUES (%s, %s)", (1, '张三'))
cur.execute("UPDATE test_user SET name = %s WHERE id = %s", ('李四', 1))
conn.commit()  # 必须提交
cur.close()
conn.close()

二、参数传递与防SQL注入

使用占位符 %s 传递参数是防止SQL注入的最佳实践。ksycopg2会自动处理参数转义，无需手动拼接字符串。

1. 位置占位符与字典占位符

当参数较多时，推荐使用字典占位符，使代码更具可读性。

# 字典传参示例
cur.execute(
    "SELECT * FROM test_user WHERE age > %(min_age)s AND name LIKE %(name_pattern)s",
    {'min_age': 18, 'name_pattern': '%张%'}
)

严禁使用 f-string 或字符串拼接直接构造SQL，例如 f"... VALUES ({id}, '{name}')"，这会带来严重的安全风险。

三、高效批量操作

批量插入数据时，不同方法的性能差异显著。选择合适的工具能大幅提升导入效率。

1. 三种批量方法对比

• executemany()：循环调用 execute()，每条SQL单独发送，网络往返次数多，性能较差。
• extras.execute_batch()：将多条SQL分组发送，减少网络往返，性能优于前者。
• extras.execute_values()：将数据拼合成一条 INSERT 语句，网络往返次数最少，性能最佳，推荐用于大批量插入。

2. 代码实现

使用 execute_values 时，SQL语句中的值部分应使用 %s 占位，并由库自动展开为多值列表。

from ksycopg2 import extras

data = [(1, '张三'), (2, '李四'), (3, '王五')]  # 假设数据量较大

# 推荐：高效批量插入
extras.execute_values(
    cur,
    "INSERT INTO test_user (id, name) VALUES %s",
    data,
    page_size=100  # 每批处理100条
)
conn.commit()

四、存储过程与函数调用

金仓数据库支持存储过程与函数。Python中调用函数可使用 callproc()，而调用存储过程则需使用 CALL 语句配合 execute()。

# 调用存储过程示例
cur.execute("CALL update_user_name(%s, %s)", (1, '新名字'))
conn.commit()

五、COPY命令：高效数据导入导出

COPY 命令是Kingbase数据库中导入导出数据最快的方式，远优于逐行 INSERT。

1. 导入数据 (copy_from)

使用 copy_from() 从文件导入数据。默认分隔符为制表符，CSV文件需指定 sep=','。

with open('data.csv', 'r') as f:
    cur.copy_from(
        file=f,
        table='test_user',
        columns=('id', 'name'),
        sep=','           # 指定分隔符
    )
conn.commit()

2. 导出数据 (copy_to)

使用 copy_to() 将数据导出到文件。若需自定义完整COPY语句（如添加表头、指定编码），可使用 copy_expert()。

copy_sql = """
    COPY test_user(id, name)
    TO STDOUT
    WITH CSV HEADER DELIMITER AS ','
"""
with open('export_with_header.csv', 'w') as f:
    cur.copy_expert(sql=copy_sql, file=f)

六、大对象 (LOB) 处理

Kingbase支持 BLOB、CLOB 和 BYTEA 类型。在Python 3中，bytea 类型返回 memoryview 对象，需调用 tobytes() 进行转换。

注意：频繁读写大对象性能较差。对于几十MB以上的文件，建议存储于文件系统，数据库中仅保存路径。

# 读取bytea数据示例
cur.execute("SELECT id, b, c, ba FROM test_lob")
rows = cur.fetchall()
for row in rows:
    for cell in row:
        if isinstance(cell, memoryview):
            print(cell[:].tobytes().decode('UTF8'), end=" ")
        else:
            print(cell, end=" ")
    print()

七、动态SQL与标识符处理

表名、字段名等标识符不能通过参数占位符传递。需使用 ksycopg2.sql 模块进行安全拼接。

from ksycopg2 import sql

# 动态构建查询
table_name = 'test_user'
id_column = 'user_id'
name_column = 'user_name'

query = sql.SQL("SELECT {id}, {name} FROM {table} WHERE {id} = %s").format(
    id=sql.Identifier(id_column),
    name=sql.Identifier(name_column),
    table=sql.Identifier(table_name)
)
cur.execute(query, (100,))

八、错误处理与事务回滚

数据库操作可能抛出多种异常，需根据类型进行针对性处理并执行回滚。

• IntegrityError：违反约束（如主键重复）。
• OperationalError：连接断开或语法错误。
• ProgrammingError：表或列不存在。

import ksycopg2
from ksycopg2 import OperationalError, IntegrityError

try:
    conn = ksycopg2.connect(database='TEST', user='SYSTEM', password='123456')
    cur = conn.cursor()
    cur.execute("INSERT INTO test_user (id, name) VALUES (%s, %s)", (1, '张三'))
    conn.commit()
except IntegrityError as e:
    print(f"违反唯一约束: {e}")
    conn.rollback()
except OperationalError as e:
    print(f"连接或操作错误: {e}")
except Exception as e:
    print(f"其他错误: {e}")
    conn.rollback()
finally:
    if 'cur' in locals(): cur.close()
    if 'conn' in locals() and conn: conn.close()

九、完整示例：从连接到批量插入

以下代码展示了从建立连接、建表、批量插入到查询统计的完整流程，涵盖了资源管理与异常处理的最佳实践。

import ksycopg2
from ksycopg2 import extras

def main():
    conn = None
    try:
        conn = ksycopg2.connect(
            database='TEST',
            user='SYSTEM',
            password='123456',
            host='127.0.0.1',
            port='54321'
        )
        conn.autocommit = False
        
        cur = conn.cursor()
        
        # 建表
        cur.execute('DROP TABLE IF EXISTS test_user')
        cur.execute('''
            CREATE TABLE test_user (
                id INTEGER PRIMARY KEY,
                name VARCHAR(100),
                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
            )
        ''')
        
        # 批量插入
        data = [(i, f'user_{i}') for i in range(1, 1001)]
        extras.execute_values(
            cur,
            "INSERT INTO test_user (id, name) VALUES %s",
            data,
            page_size=200
        )
        
        # 查询验证
        cur.execute("SELECT COUNT(*) FROM test_user")
        count = cur.fetchone()[0]
        print(f"插入了 {count} 条数据")
        
        conn.commit()
        cur.close()
        
    except Exception as e:
        print(f"操作失败: {e}")
        if conn:
            conn.rollback()
    finally:
        if conn:
            conn.close()

if __name__ == "__main__":
    main()

十、总结

掌握 ksycopg2 的核心功能，包括 execute() 执行、结果集获取、参数防注入、execute_values 批量插入、COPY 命令高效导入以及大对象处理，是提升Python与金仓数据库交互效率的关键。同时，合理的错误处理与事务回滚机制能确保数据的一致性与系统的稳定性。

以上就是Python金仓数据库操作的详细内容，更多关于Python数据库交互、ksycopg2使用技巧及金仓数据库优化的资料请关注本站其它相关文章！