最新国产好看的视频,伊人天堂AV在线,国产Aaaaaa视频,蜜臀视频在线观看一区,人妻av色图,密臀久久久精品影片,青青视频免费观看毛片,久草在线观看视,国产三级精品色情在线

Python金倉數(shù)據(jù)庫操作之SQL執(zhí)行,批量操作與擴展功能的完全指南

 更新時間：2026年04月30日 08:26:14   作者：碼農(nóng)阿豪@新空間  
本文介紹了Python中使用ksycopg2操作金倉(Kingbase)數(shù)據(jù)庫的高級技巧,主要內(nèi)容包括,SQL執(zhí)行與結果處理,基礎查詢流程與游標使用,感興趣的小伙伴可以跟隨小編一起學習一下

接上篇

上篇講了 ksycopg2 的安裝配置、連接管理、高可用和連接池。這篇接著說怎么執(zhí)行 SQL、處理結果集，以及批量操作、COPY 命令這些實用功能。

一、執(zhí)行 SQL 語句

1.1 基礎查詢

先看一個完整的查詢流程：

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,))

# 獲取結果
rows = cur.fetchall()
for row in rows:
    print(f"id: {row[0]}, name: {row[1]}")

cur.close()
conn.close()

幾個要點：

  • 先用 cursor() 創(chuàng)建游標
  • execute() 執(zhí)行 SQL
  • fetchall() 拿全部結果
  • 用完記得關閉游標和連接

1.2 獲取結果集的幾種方式

fetchone()：一條一條拿

cur.execute("SELECT id, name FROM test_user")
while True:
    row = cur.fetchone()
    if row is None:
        break
    print(row)

適合處理大結果集，不會一次性把所有數(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 快，又不會像 fetchall 那樣吃內(nèi)存。

fetchall()：一次性全拿

cur.execute("SELECT id, name FROM test_user")
rows = cur.fetchall()  # 數(shù)據(jù)量小的時候用
for row in rows:
    print(row)

數(shù)據(jù)量小的時候方便，幾萬條以上就別用了。

1.3 獲取列信息

有時候需要知道查詢結果有哪些列、什么類型，可以用 cursor.description：

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}")

輸出示例：

列名: id, 類型碼: 23, 長度: 4
列名: name, 類型碼: 1043, 長度: -1
列名: created_at, 類型碼: 1114, 長度: 8

類型碼是 Kingbase 內(nèi)部的數(shù)據(jù)類型 OID，一般用不到，但調(diào)試時有用。

1.4 執(zhí)行非查詢 SQL

INSERT、UPDATE、DELETE 這些不返回結果集的 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,))

# 注意：需要提交事務
conn.commit()
cur.close()

千萬別忘了 commit()，否則數(shù)據(jù)不會真正寫入。

二、參數(shù)傳遞與防 SQL 注入

ksycopg2 用占位符 %s 傳遞參數(shù)，會自動處理轉(zhuǎn)義，不用自己拼接字符串。

# 正確寫法：參數(shù)單獨傳遞
cur.execute(
    "INSERT INTO test_user (id, name) VALUES (%s, %s)",
    (1, "張三")
)

# 錯誤寫法：自己拼接字符串，有 SQL 注入風險
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ù)多的時候更清晰
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() 的問題

ksycopg2 提供了 executemany()，但它的實現(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ù)量小的時候用用還行，大批量插入不建議。

3.2 execute_batch() 批量執(zhí)行

ksycopg2.extras.execute_batch() 把多條 SQL 分成若干組，每組一次發(fā)給數(shù)據(jù)庫，減少網(wǎng)絡往返次數(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)絡交互多，太大 SQL 語句太長，100-500 之間比較合適。

3.3 execute_values() 一條 SQL 插多行

execute_values() 把所有數(shù)據(jù)拼成一條 INSERT 語句，效率最高：

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ù)時，execute_values 比 execute_batch 快不少。

三種批量插入方式對比（插入 1 萬條數(shù)據(jù)測試）：

方式網(wǎng)絡往返速度適用場景
executemany1萬次慢少量數(shù)據(jù)
execute_batch100次中中等數(shù)據(jù)量
execute_values1次快大批量數(shù)據(jù)

四、調(diào)用存儲過程

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)用存儲過程

調(diào)用存儲過程需要用 execute() 配合 CALL 語句：

-- 創(chuàng)建存儲過程
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)用存儲過程
cur.execute("CALL update_user_name(%s, %s)", (1, '新名字'))
conn.commit()

五、COPY 命令：高效數(shù)據(jù)導入導出

COPY 是 Kingbase 提供的快速數(shù)據(jù)導入導出方式，比 INSERT 快很多。

5.1 copy_from()：從文件導入

cur = conn.cursor()

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

默認分隔符是制表符 \t，CSV 文件需要指定 sep=','。NULL 值默認用 \N 表示，也可以改：

cur.copy_from(f, 'test_user', columns=('id', 'name'), sep=',', null='NULL')

5.2 copy_to()：導出到文件

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 語句：

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ù)遷移、日志導出、報表生成這些場景，效率比 SELECT 一行行寫高得多。

六、大對象處理

Kingbase 支持 BLOB（二進制大對象）和 CLOB（字符大對象）。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
    )
''')

# 準備測試數(shù)據(jù)
ba_data = bytearray("中文測試bytearray", "UTF8")
b_data = bytes('中文測試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()

三個注意事項：

  • bytea 類型在 Python 3 里返回 memoryview，用 tobytes() 轉(zhuǎn)換
  • 大對象不建議頻繁讀寫，性能不好
  • 超大文件（幾十 MB 以上）建議存文件系統(tǒng)，數(shù)據(jù)庫只存路徑

七、動態(tài) SQL 生成

ksycopg2.sql 模塊解決了一個頭疼的問題：表名、字段名這類標識符不能直接用參數(shù)傳遞。

from ksycopg2 import sql

# 錯誤寫法：標識符不能參數(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 拼接帶標識符的 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 動態(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
)

八、錯誤處理

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:
    cur.close()
    conn.close()

常見異常類型：

  • IntegrityError：違反約束（主鍵重復、外鍵不存在等）
  • OperationalError：連接斷開、語法錯誤等
  • 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()

十、總結

下篇主要講了這幾個方面：

  1. SQL 執(zhí)行：execute() 執(zhí)行查詢和非查詢語句，fetchone/fetchmany/fetchall 獲取結果
  2. 參數(shù)傳遞：用 %s 占位符，不要拼接字符串
  3. 批量操作：execute_values 性能最佳，大批量插入首選
  4. COPY 命令：數(shù)據(jù)導入導出最快的方式
  5. 大對象：BLOB/CLOB/BYTEA 的處理方式
  6. 動態(tài) SQL：表名、字段名用 sql 模塊處理
  7. 錯誤處理：區(qū)分異常類型，正確處理事務回滾

兩篇合在一起，覆蓋了 Python 操作金倉數(shù)據(jù)庫的常用場景。從連接到高可用，從 SQL 執(zhí)行到批量操作，基本夠用了。如果遇到官方文檔沒覆蓋的場景，建議去金倉官網(wǎng)看最新的驅(qū)動包和示例。

到此這篇關于Python金倉數(shù)據(jù)庫操作之SQL執(zhí)行,批量操作與擴展功能的完全指南的文章就介紹到這了,更多相關Python操作金倉數(shù)據(jù)庫內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關文章希望大家以后多多支持腳本之家！

相關文章

最新評論

五指山市| 嘉兴市| 长白| 霞浦县| 徐州市| 乌兰县| 西丰县| 崇明县| 永城市| 若尔盖县| 北票市| 乡宁县| 南安市| 甘泉县| 紫阳县| 宜川县| 玉门市| 寻甸| 建宁县| 司法| 法库县| 张家川| 安宁市| 贵德县| 保靖县| 大竹县| 永定县| 马关县| 全州县| 天门市| 武汉市| 依安县| 通河县| 徐州市| 防城港市| 聊城市| 彭山县| 大邑县| 丰城市| 上杭县| 高安市|