Python金倉(cāng)數(shù)據(jù)庫(kù)操作之SQL執(zhí)行,批量操作與擴(kuò)展功能的完全指南
更新時(shí)間：2026年04月30日 08:26:14 作者：碼農(nóng)阿豪@新空間
本文介紹了Python中使用ksycopg2操作金倉(cāng)(Kingbase)數(shù)據(jù)庫(kù)的高級(jí)技巧,主要內(nèi)容包括,SQL執(zhí)行與結(jié)果處理,基礎(chǔ)查詢流程與游標(biāo)使用,感興趣的小伙伴可以跟隨小編一起學(xué)習(xí)一下
接上篇
上篇講了 ksycopg2 的安裝配置、連接管理、高可用和連接池。這篇接著說(shuō)怎么執(zhí)行 SQL、處理結(jié)果集，以及批量操作、COPY 命令這些實(shí)用功能。
一、執(zhí)行 SQL 語(yǔ)句
1.1 基礎(chǔ)查詢
先看一個(gè)完整的查詢流程：
import ksycopg2
conn = ksycopg2.connect(
database='TEST',
user='SYSTEM',
password='123456',
host='127.0.0.1',
port='54321'
)
cur = conn.cursor()
# 執(zhí)行查詢
cur.execute("SELECT id, name FROM test_user WHERE age > %s", (18,))
# 獲取結(jié)果
rows = cur.fetchall()
for row in rows:
print(f"id: {row[0]}, name: {row[1]}")
cur.close()
conn.close()
幾個(gè)要點(diǎn)：
- 先用
cursor()創(chuàng)建游標(biāo) execute()執(zhí)行 SQLfetchall()拿全部結(jié)果- 用完記得關(guān)閉游標(biāo)和連接
1.2 獲取結(jié)果集的幾種方式
fetchone()：一條一條拿
cur.execute("SELECT id, name FROM test_user")
while True:
row = cur.fetchone()
if row is None:
break
print(row)
適合處理大結(jié)果集，不會(huì)一次性把所有數(shù)據(jù)加載到內(nèi)存。
fetchmany()：分批拿
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)
比 fetchone 快，又不會(huì)像 fetchall 那樣吃內(nèi)存。
fetchall()：一次性全拿
cur.execute("SELECT id, name FROM test_user")
rows = cur.fetchall() # 數(shù)據(jù)量小的時(shí)候用
for row in rows:
print(row)
數(shù)據(jù)量小的時(shí)候方便，幾萬(wàn)條以上就別用了。
1.3 獲取列信息
有時(shí)候需要知道查詢結(jié)果有哪些列、什么類型，可以用 cursor.description：
cur.execute("SELECT id, name, created_at FROM test_user")
for col in cur.description:
print(f"列名: {col.name}, 類型碼: {col.type_code}, 長(zhǎng)度: {col.internal_size}")
輸出示例：
列名: id, 類型碼: 23, 長(zhǎng)度: 4
列名: name, 類型碼: 1043, 長(zhǎng)度: -1
列名: created_at, 類型碼: 1114, 長(zhǎng)度: 8
類型碼是 Kingbase 內(nèi)部的數(shù)據(jù)類型 OID，一般用不到，但調(diào)試時(shí)有用。
1.4 執(zhí)行非查詢 SQL
INSERT、UPDATE、DELETE 這些不返回結(jié)果集的 SQL，直接用 execute() 執(zhí)行就行：
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))
# 刪除
cur.execute("DELETE FROM test_user WHERE id = %s", (1,))
# 注意：需要提交事務(wù)
conn.commit()
cur.close()
千萬(wàn)別忘了 commit()，否則數(shù)據(jù)不會(huì)真正寫入。
二、參數(shù)傳遞與防 SQL 注入
ksycopg2 用占位符 %s 傳遞參數(shù)，會(huì)自動(dòng)處理轉(zhuǎn)義，不用自己拼接字符串。
# 正確寫法：參數(shù)單獨(dú)傳遞
cur.execute(
"INSERT INTO test_user (id, name) VALUES (%s, %s)",
(1, "張三")
)
# 錯(cuò)誤寫法：自己拼接字符串，有 SQL 注入風(fēng)險(xiǎn)
cur.execute(f"INSERT INTO test_user (id, name) VALUES ({id}, '{name}')")
2.1 位置占位符
# 用元組傳參
cur.execute(
"SELECT * FROM test_user WHERE age > %s AND name LIKE %s",
(18, '%張%')
)
2.2 命名占位符
# 用字典傳參，參數(shù)多的時(shí)候更清晰
cur.execute(
"SELECT * FROM test_user WHERE age > %(min_age)s AND name LIKE %(name_pattern)s",
{'min_age': 18, 'name_pattern': '%張%'}
)
三、批量操作
3.1 executemany() 的問(wèn)題
ksycopg2 提供了 executemany()，但它的實(shí)現(xiàn)方式是循環(huán)調(diào)用 execute()，一次發(fā)一條 SQL，性能提升不大。
data = [(1, '張三'), (2, '李四'), (3, '王五')]
cur.executemany("INSERT INTO test_user (id, name) VALUES (%s, %s)", data)
conn.commit()
數(shù)據(jù)量小的時(shí)候用用還行，大批量插入不建議。
3.2 execute_batch() 批量執(zhí)行
ksycopg2.extras.execute_batch() 把多條 SQL 分成若干組，每組一次發(fā)給數(shù)據(jù)庫(kù)，減少網(wǎng)絡(luò)往返次數(shù)。
from ksycopg2 import extras
data = [(1, '張三'), (2, '李四'), (3, '王五'), ...] # 幾百條數(shù)據(jù)
extras.execute_batch(
cur,
"INSERT INTO test_user (id, name) VALUES (%s, %s)",
data,
page_size=100 # 每100條發(fā)一次
)
conn.commit()
page_size 控制每組多少條。太小網(wǎng)絡(luò)交互多，太大 SQL 語(yǔ)句太長(zhǎng)，100-500 之間比較合適。
3.3 execute_values() 一條 SQL 插多行
execute_values() 把所有數(shù)據(jù)拼成一條 INSERT 語(yǔ)句，效率最高：
from ksycopg2 import extras
data = [(1, '張三'), (2, '李四'), (3, '王五')]
extras.execute_values(
cur,
"INSERT INTO test_user (id, name) VALUES %s",
data,
page_size=100
)
conn.commit()
生成的 SQL 類似：
INSERT INTO test_user (id, name) VALUES (1, '張三'), (2, '李四'), (3, '王五')
一次性插入幾千條數(shù)據(jù)時(shí)，execute_values 比 execute_batch 快不少。
三種批量插入方式對(duì)比（插入 1 萬(wàn)條數(shù)據(jù)測(cè)試）：
| 方式 | 網(wǎng)絡(luò)往返 | 速度 | 適用場(chǎng)景 |
|---|---|---|---|
| executemany | 1萬(wàn)次 | 慢 | 少量數(shù)據(jù) |
| execute_batch | 100次 | 中 | 中等數(shù)據(jù)量 |
| execute_values | 1次 | 快 | 大批量數(shù)據(jù) |
四、調(diào)用存儲(chǔ)過(guò)程
4.1 調(diào)用函數(shù)
# 先創(chuàng)建函數(shù)
cur.execute("""
CREATE OR REPLACE FUNCTION add_user(p_id INTEGER, p_name TEXT)
RETURNS TEXT AS $$
BEGIN
INSERT INTO test_user (id, name) VALUES (p_id, p_name);
RETURN 'success';
END;
$$ LANGUAGE plpgsql;
""")
# 調(diào)用函數(shù)
cur.callproc('add_user', (10, 'test_user'))
result = cur.fetchone()
print(result) # ('success',)
conn.commit()
4.2 調(diào)用存儲(chǔ)過(guò)程
調(diào)用存儲(chǔ)過(guò)程需要用 execute() 配合 CALL 語(yǔ)句：
-- 創(chuàng)建存儲(chǔ)過(guò)程
CREATE OR REPLACE PROCEDURE update_user_name(
p_id INTEGER,
p_new_name TEXT
)
LANGUAGE plpgsql
AS $$
BEGIN
UPDATE test_user SET name = p_new_name WHERE id = p_id;
END;
$$;# 調(diào)用存儲(chǔ)過(guò)程
cur.execute("CALL update_user_name(%s, %s)", (1, '新名字'))
conn.commit()
五、COPY 命令：高效數(shù)據(jù)導(dǎo)入導(dǎo)出
COPY 是 Kingbase 提供的快速數(shù)據(jù)導(dǎo)入導(dǎo)出方式，比 INSERT 快很多。
5.1 copy_from()：從文件導(dǎo)入
cur = conn.cursor()
with open('data.csv', 'r') as f:
cur.copy_from(
file=f,
table='test_user',
columns=('id', 'name'),
sep=',' # 列分隔符
)
conn.commit()
默認(rèn)分隔符是制表符 \t，CSV 文件需要指定 sep=','。NULL 值默認(rèn)用 \N 表示，也可以改：
cur.copy_from(f, 'test_user', columns=('id', 'name'), sep=',', null='NULL')
5.2 copy_to()：導(dǎo)出到文件
with open('export.csv', 'w') as f:
cur.copy_to(
file=f,
table='test_user',
columns=('id', 'name'),
sep=','
)
conn.commit()
5.3 copy_expert()：自定義 COPY
copy_expert 最靈活，可以寫完整的 COPY 語(yǔ)句：
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)
用在數(shù)據(jù)遷移、日志導(dǎo)出、報(bào)表生成這些場(chǎng)景，效率比 SELECT 一行行寫高得多。
六、大對(duì)象處理
Kingbase 支持 BLOB（二進(jìn)制大對(duì)象）和 CLOB（字符大對(duì)象）。ksycopg2 可以處理這些類型。
import ksycopg2
conn = ksycopg2.connect(
database='TEST', user='SYSTEM',
password='123456', host='127.0.0.1', port='54321'
)
cur = conn.cursor()
# 建表
cur.execute('DROP TABLE IF EXISTS test_lob')
cur.execute('''
CREATE TABLE test_lob (
id INTEGER,
b BLOB,
c CLOB,
ba BYTEA
)
''')
# 準(zhǔn)備測(cè)試數(shù)據(jù)
ba_data = bytearray("中文測(cè)試bytearray", "UTF8")
b_data = bytes('中文測(cè)試bytes' * 2, "UTF8")
str_data = '中文str' * 4
# 插入
cur.execute(
"INSERT INTO test_lob VALUES (%s, %s, %s, %s)",
(1, ba_data, ba_data, ba_data)
)
cur.execute(
"INSERT INTO test_lob VALUES (%s, %s, %s, %s)",
(2, b_data, b_data, b_data)
)
cur.execute(
"INSERT INTO test_lob VALUES (%s, %s, %s, %s)",
(3, str_data, str_data, str_data)
)
conn.commit()
# 讀取
cur.execute("SELECT id, b, c, ba FROM test_lob")
rows = cur.fetchall()
for row in rows:
for cell in row:
# bytea 類型返回 memoryview，需要轉(zhuǎn)換
if isinstance(cell, memoryview):
print(cell[:].tobytes().decode('UTF8'), end=" ")
else:
print(cell, end=" ")
print()
cur.close()
conn.close()
三個(gè)注意事項(xiàng)：
bytea類型在 Python 3 里返回memoryview，用tobytes()轉(zhuǎn)換- 大對(duì)象不建議頻繁讀寫，性能不好
- 超大文件（幾十 MB 以上）建議存文件系統(tǒng)，數(shù)據(jù)庫(kù)只存路徑
七、動(dòng)態(tài) SQL 生成
ksycopg2.sql 模塊解決了一個(gè)頭疼的問(wèn)題：表名、字段名這類標(biāo)識(shí)符不能直接用參數(shù)傳遞。
from ksycopg2 import sql
# 錯(cuò)誤寫法：標(biāo)識(shí)符不能參數(shù)化
cur.execute("SELECT * FROM %s WHERE id = %s", ('test_user', 1))
# 正確寫法：用 sql 模塊
cur.execute(
sql.SQL("SELECT * FROM {} WHERE id = %s").format(sql.Identifier('test_user')),
(1,)
)
7.1 拼接帶標(biāo)識(shí)符的 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,))
7.2 使用字面值
from ksycopg2 import sql
# 把 Python 值直接轉(zhuǎn)成 SQL 字面量
value = sql.Literal("It's a test")
cur.execute(
sql.SQL("SELECT {}").format(value)
)
# 生成: SELECT 'It''s a test'
7.3 動(dòng)態(tài) IN 查詢
ids = [1, 2, 3, 4, 5]
placeholders = ','.join(['%s'] * len(ids))
cur.execute(
f"SELECT * FROM test_user WHERE id IN ({placeholders})",
ids
)
八、錯(cuò)誤處理
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"連接或操作錯(cuò)誤: {e}")
except Exception as e:
print(f"其他錯(cuò)誤: {e}")
conn.rollback()
finally:
cur.close()
conn.close()
常見(jiàn)異常類型：
IntegrityError：違反約束（主鍵重復(fù)、外鍵不存在等）OperationalError：連接斷開(kāi)、語(yǔ)法錯(cuò)誤等ProgrammingError：表不存在、列不存在等
九、完整示例：從連接到批量插入
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} 條數(shù)據(jù)")
conn.commit()
cur.close()
except Exception as e:
print(f"操作失敗: {e}")
if conn:
conn.rollback()
finally:
if conn:
conn.close()
if __name__ == "__main__":
main()
十、總結(jié)
下篇主要講了這幾個(gè)方面：
- SQL 執(zhí)行：
execute()執(zhí)行查詢和非查詢語(yǔ)句，fetchone/fetchmany/fetchall獲取結(jié)果 - 參數(shù)傳遞：用
%s占位符，不要拼接字符串 - 批量操作：
execute_values性能最佳，大批量插入首選 - COPY 命令：數(shù)據(jù)導(dǎo)入導(dǎo)出最快的方式
- 大對(duì)象：BLOB/CLOB/BYTEA 的處理方式
- 動(dòng)態(tài) SQL：表名、字段名用
sql模塊處理 - 錯(cuò)誤處理：區(qū)分異常類型，正確處理事務(wù)回滾
兩篇合在一起，覆蓋了 Python 操作金倉(cāng)數(shù)據(jù)庫(kù)的常用場(chǎng)景。從連接到高可用，從 SQL 執(zhí)行到批量操作，基本夠用了。如果遇到官方文檔沒(méi)覆蓋的場(chǎng)景，建議去金倉(cāng)官網(wǎng)看最新的驅(qū)動(dòng)包和示例。
到此這篇關(guān)于Python金倉(cāng)數(shù)據(jù)庫(kù)操作之SQL執(zhí)行,批量操作與擴(kuò)展功能的完全指南的文章就介紹到這了,更多相關(guān)Python操作金倉(cāng)數(shù)據(jù)庫(kù)內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！
相關(guān)文章
Python數(shù)據(jù)結(jié)構(gòu)之循環(huán)鏈表詳解
循環(huán)鏈表 (Circular Linked List) 是鏈?zhǔn)酱鎯?chǔ)結(jié)構(gòu)的另一種形式，它將鏈表中最后一個(gè)結(jié)點(diǎn)的指針指向鏈表的頭結(jié)點(diǎn)，使整個(gè)鏈表頭尾相接形成一個(gè)環(huán)形，使鏈表的操作更加方便靈活。本文將詳細(xì)介紹一下循環(huán)鏈表的相關(guān)知識(shí)，需要的可以參考一下2022-01-01
Python 快速把多個(gè)元素連接成一個(gè)字符串的操作方法
join() 方法一個(gè)用于將序列中的元素以指定的分隔符連接成一個(gè)字符串的方法,這個(gè)方法通常用于字符串操作,這篇文章主要介紹了Python 快速把多個(gè)元素連接成一個(gè)字符串的方法,需要的朋友可以參考下2024-06-06
全面解析Python的While循環(huán)語(yǔ)句的使用方法
這篇文章主要介紹了全面解析Python的While循環(huán)語(yǔ)句的使用方法,是Python入門學(xué)習(xí)中的基礎(chǔ)知識(shí),需要的朋友可以參考下2015-10-10
這篇文章主要為大家詳細(xì)介紹了python實(shí)現(xiàn)梯度下降算法，文中示例代碼介紹的非常詳細(xì)，具有一定的參考價(jià)值，感興趣的小伙伴們可以參考一下2018-08-08
Python2.5/2.6實(shí)用教程 入門基礎(chǔ)篇
本文方便有經(jīng)驗(yàn)的程序員進(jìn)入Python世界.本文適用于python2.5/2.6版本.2009-11-11
詳解python之多進(jìn)程和進(jìn)程池(Processing庫(kù))
本篇文章主要介紹了詳解python之多進(jìn)程和進(jìn)程池(Processing庫(kù))，非常具有實(shí)用價(jià)值，需要的朋友可以參考下2017-06-06
這篇文章主要為大家詳細(xì)介紹了python微信好友數(shù)據(jù)分析，實(shí)現(xiàn)對(duì)微信好友的獲取，并對(duì)省份、性別等數(shù)據(jù)分析，具有一定的參考價(jià)值，感興趣的小伙伴們可以參考一下2018-11-11
舉例講解Python設(shè)計(jì)模式編程中的訪問(wèn)者與觀察者模式
這篇文章主要介紹了Python設(shè)計(jì)模式編程中的訪問(wèn)者與觀察者模式,設(shè)計(jì)模式的制定有利于團(tuán)隊(duì)協(xié)作編程代碼的協(xié)調(diào),需要的朋友可以參考下2016-01-01
Python中常用操作字符串的函數(shù)與方法總結(jié)
這篇文章主要介紹了Python中常用操作字符串的函數(shù)與方法總結(jié),包括字符串的格式化輸出與拼接等基礎(chǔ)知識(shí),需要的朋友可以參考下2016-02-02
windows環(huán)境下python程序庫(kù)導(dǎo)出requirements并使用詳解
這篇文章主要介紹了windows環(huán)境下python程序庫(kù)導(dǎo)出requirements并使用方式,具有很好的參考價(jià)值,希望對(duì)大家有所幫助,如有錯(cuò)誤或未考慮完全的地方,望不吝賜教2025-05-05