Python 使用 Ksycopg2 连接和操作 Kingbase 数据库实战

创建游标：

cur = conn.cursor()

执行 SQL 语句（建表、插入数据）：

cur.execute('drop table if exists test_ksy')
cur.execute('create table test_ksy(id integer, name TEXT)')
cur.execute("insert into test_ksy values(%s, %s)", (1, "John"))
cur.execute("insert into test_ksy values(%s, %s)", (2, '中文测试文字'))
cur.execute("insert into test_ksy values(%s, %s)", (3, '！@#￥%……'))

查询并获取结果集：

cur.execute("select * from test_ksy")
rows = cur.fetchall()
print("result for query:")
for row in rows:
    print("\t", row)

关闭游标及连接：

cur.close()
conn.close()

完整示例：

# -*- coding: utf-8 -*-
import ksycopg2

database = "TEST"
user = "SYSTEM"
password = "123456"
host = "127.0.0.1"
port = "54321"

try:
    conn = ksycopg2.connect("dbname={} user={} password={} host={} port={}".format(database, user, password, host, port))
    cur = conn.cursor()
    cur.execute('drop table if exists test_ksy')
    cur.execute('create table test_ksy(id integer, name TEXT)')
    cur.execute("insert into test_ksy values(%s, %s)", (1, "John"))
    cur.execute("insert into test_ksy values(%s, %s)", (2, '中文测试文字'))
    cur.execute("insert into test_ksy values(%s, %s)", (3, '！@#￥%……'))
    cur.execute("select * from test_ksy")
    rows = cur.fetchall()
    print(type(rows), type(rows[0]))
    print("result for query:")
    for row in rows:
        print("\t", row)
    cur.close()
    conn.close()
except Exception as e:
    print(e)
    exit()

四、运行程序

执行脚本：

python test_ksycopg2.py

输出示例：

<class 'list'><class 'tuple'> result for query:
(1,'John')
(2,'中文测试文字')
(3,'！@#￥%……')

五、常见 Bug 处理

1. SSL 库冲突

   • 原因：系统环境 SSL 库版本过高或加载了错误的库。
   • 解决：执行 ldd _ksycopg 检查依赖。若仍报错，联系技术支持获取包含静态依赖 SSL 库的驱动包。

2. 模块加载失败

   • 报错：No module named 'ksycopg2._ksycopg'
   • 原因：架构不匹配、Python 版本冲突或路径错误。
   • 解决：确保安装匹配的驱动包。若已安装但仍报错，可强制设置模块搜索路径：

     import sys
     sys.path.insert(0, "Ksycopg2 驱动包的父目录")
     

六、结语

掌握 Python 与 Kingbase 的集成技术，有助于实现国产数据库的应用开发。后续可进一步探索其他语言连接方案。