我要评论开篇：为什么需要这篇指南
去年做个数据分析平台，后端用python开发，要连金仓数据库做数据清洗和统计。一开始图省事，用了普通数据库驱动，结果遇到一堆麻烦：
- 代码到处都是
try-catch-finally，看着都烦 - 并发一上来，连接动不动就超时
- 最要命的是数据库主备切换时，应用直接挂了
后来换成了ksycopg2驱动，这些问题基本都解决了。这篇分上下两篇，上篇说说连接管理、高可用配置和数据类型映射，下篇讲sql执行、批量操作和扩展功能。
一、ksycopg2 是什么
ksycopg2是金仓数据库的python驱动，用c语言写的，性能不错，完整实现了python db api 2.0规范。说白了，就是python连接金仓数据库的标准方式。
它依赖libkci库，libkci又依赖openssl。所以安装时要是报缺库，多半是这两个依赖没配好。
二、安装配置
2.1 版本选择
ksycopg2的版本号得和python版本对上。比如包名是ksycopg2‑python2.7，就只能用在python 2.7上。
官方支持的情况：
| python 版本 | linux 支持 | windows 支持 |
|---|---|---|
| python 2.7 | x86_64、arm、loongarch、mips、sw | 64位 |
| python 3.5-3.13 | x86_64、arm、loongarch、mips | 64位 |
我自己用的是python 3.9，对应的驱动跑得挺稳。
2.2 安装步骤
最简单的就是用pip装。不过得先确认pip对应哪个python版本。
# 查看 pip 对应的 python 版本 pip --version # 安装 ksycopg2 pip install ksycopg2
要是pip装不了，或者要离线安装，就去金仓官网下载对应版本的驱动包，解压后把ksycopg2文件夹扔到python的site-packages目录下。
怎么找这个目录？在python环境里执行：
import sys sys.path
输出的列表里，带site-packages的那个路径就是了。
2.3 依赖库配置
ksycopg2依赖libkci.so（linux）或libkci.dll（windows）。驱动装好了但导入报错，通常是没找到这些库。
报错信息长这样：
importerror: libkci.so: cannot open shared object file: no such file or directory
解决办法：把libkci所在目录加到ld_library_path环境变量里。
export ld_library_path=/home/kingbase/lib:$ld_library_path
windows上就把libkci.dll放到python安装目录或系统path里。
一个常见坑：windows下导入ksycopg2时报dll load failed，多半是缺msvc运行库。装上visual c++ redistributable就好了。
2.4 验证安装
import ksycopg2 print(ksycopg2.__version__)
不报错且输出版本号，说明装好了。
三、连接数据库
3.1 三种连接方式
方式一：键值对字符串
import ksycopg2
conn = ksycopg2.connect(
"dbname=test user=system password=123456 host=127.0.0.1 port=54321"
)
方式二：关键字参数
这种方式代码更清晰，生产环境推荐用：
conn = ksycopg2.connect(
database='test',
user='system',
password='123456',
host='127.0.0.1',
port='54321',
client_encoding='utf8' # 设置客户端编码
)
方式三：uri 格式
conn = ksycopg2.connect(
"kingbase://system:123456@127.0.0.1:54321/test"
)
还有一种uri格式：
conn = ksycopg2.connect(
"postgresql://system:123456@127.0.0.1:54321/test"
)
支持postgresql前缀主要是历史原因，实际连的还是金仓。
3.2 连接参数说明
| 参数 | 说明 | 默认值 |
|---|---|---|
| host | 服务器地址 | localhost |
| port | 端口 | 54321 |
| user | 用户名 | 当前系统用户 |
| password | 密码 | 无 |
| dbname | 数据库名 | 同用户名 |
| connect_timeout | 连接超时秒数 | 无限等待 |
| client_encoding | 客户端编码 | 数据库默认编码 |
| sslmode | ssl 模式 | prefer |
其中connect_timeout建议设一下，默认无限等待在某些网络环境下会卡住。
3.3 关闭连接
用完记得关，不然会占着数据库连接数：
conn.close()
可以用with语句自动管理资源：
with ksycopg2.connect(database='test', user='system') as conn:
with conn.cursor() as cur:
cur.execute("select 1")
print(cur.fetchone())
# 退出 with 块时自动提交事务，但连接不会自动关闭，还得手动 conn.close()
四、高可用配置（重点）
这是ksycopg2比较实用的功能之一。如果你的金仓数据库是集群部署（主备模式），配好高可用参数，主备切换时应用就能自动重连。
4.1 多主机配置
先看基础的多主机配置：
# 配置两个主机地址
hosts = "192.168.1.100,192.168.1.101"
ports = "54321"
conn = ksycopg2.connect(
f"dbname=test user=system password=123456 host={hosts} port={ports}"
)
驱动会按顺序尝试连接。第一个连不上，自动切到第二个。
4.2 连接重试
集群主备切换时，数据库服务会短暂不可用。配置重试参数可以避免应用直接报错：
conn = ksycopg2.connect(
f"dbname=test user=system password=123456 host={hosts} port={ports} "
"connect_timeout=3 retries=5 delay=3"
)
参数说明：
retries=5：最多重试5次delay=3：每次重试间隔3秒connect_timeout=3：每次连接尝试最长等3秒
这么配下来，数据库最长需要等多久？实际逻辑是：第1次连接3秒超时，等3秒，第2次连接3秒超时，等3秒……主备切换通常10-15秒能完成，5次重试够了。
4.3 自动找主
集群环境里，只有主节点支持读写。可以配置target_session_attrs=read-write，让驱动自动找到主节点连接：
conn = ksycopg2.connect(
f"dbname=test user=system password=123456 host={hosts} port={ports} "
"connect_timeout=3 target_session_attrs=read-write"
)
驱动会逐个主机尝试，直到连到一个支持读写的节点。这个配置在读写分离场景下特别有用——应用只管写，不需要知道当前谁是主。
4.4 快速故障转移
fastswitch=on是另一个实用参数：
conn = ksycopg2.connect(
f"dbname=test user=system password=123456 host={hosts} port={ports} "
"fastswitch=on retries=3 delay=5"
)
开启后，驱动会记住哪些主机挂了，后续连接请求自动跳过故障主机，不用每次都等超时。对于短连接频繁的场景，这个参数能明显减少延迟。
参数对比：
| 参数 | 作用 | 适用场景 |
|---|---|---|
| retries/delay | 连接失败时重试 | 主备切换、网络抖动 |
| target_session_attrs | 只连主节点 | 读写分离、写操作 |
| fastswitch | 跳过故障主机 | 短连接、高并发 |
4.5 连接负载均衡
如果后端有多个备节点，可以开启负载均衡分散连接压力：
conn = ksycopg2.connect(
f"dbname=test user=system password=123456 host={hosts} port={ports} "
"loadbalance=on connect_timeout=6"
)
参数说明：
loadbalance=on：开启负载均衡- 驱动会在多个主机之间轮询分配连接
- 适合读多写少、有多个备节点的场景
五、连接池
5.1 simpleconnectionpool（单线程）
单线程场景（比如定时任务脚本），用simpleconnectionpool：
import ksycopg2
from ksycopg2 import pool
connection_pool = ksycopg2.pool.simpleconnectionpool(
minconn=1,
maxconn=10,
database='test',
user='system',
password='123456',
host='127.0.0.1',
port='54321'
)
# 获取连接
conn = connection_pool.getconn()
cur = conn.cursor()
cur.execute("select * from test_user")
rows = cur.fetchall()
cur.close()
# 放回连接池
connection_pool.putconn(conn)
# 关闭所有连接
connection_pool.closeall()
5.2 threadedconnectionpool（多线程）
多线程web应用（比如flask、django部署），建议用threadedconnectionpool：
connection_pool = ksycopg2.pool.threadedconnectionpool(
minconn=5,
maxconn=20,
database='test',
user='system',
password='123456',
host='127.0.0.1',
port='54321'
)
使用方法跟simpleconnectionpool一样。
一个需要注意的地方：putconn()只是把连接放回池子里，不会关闭连接。多线程环境下，如果多个线程同时拿到同一个连接，可能会出问题。需要自己在应用层加锁控制。
另一个常见的错误：千万不要对getconn()拿到的连接直接调用close()。那样连接池里的连接数就少了一个，最后会报connection pool exhausted。
如果连接池满了，getconn()会直接报错。解决办法是调大maxconn，或者排查业务代码有没有及时putconn()。
5.3 完整的多线程示例
import threading
import ksycopg2
from ksycopg2 import pool
connection_pool = ksycopg2.pool.threadedconnectionpool(
1, 5, database='test', user='system',
password='123456', host='127.0.0.1', port='54321'
)
def worker(n):
conn = connection_pool.getconn()
cur = conn.cursor()
cur.execute("insert into test_user values (%s, %s)", (n, f"user_{n}"))
conn.commit()
cur.close()
connection_pool.putconn(conn)
threads = []
for i in range(10):
t = threading.thread(target=worker, args=(i,))
threads.append(t)
t.start()
for t in threads:
t.join()
connection_pool.closeall()
六、自动提交模式
ksycopg2默认是手动提交模式。执行insert、update后必须调用commit()，否则数据不会真正写入。
conn = ksycopg2.connect(database='test', user='system')
conn.autocommit = false # 默认值
cur = conn.cursor()
cur.execute("insert into test_user values (1, 'test')")
conn.commit() # 必须调用
如果不想手动提交，可以开启自动提交模式：
conn.autocommit = true
开启后，每条语句执行完立刻提交。适合原型开发，生产环境不建议用，事务控制会乱。
七、数据类型映射
python类型和金仓数据库类型的对应关系：
| kingbase 类型 | python 类型 |
|---|---|
| null | none |
| smallint, integer, bigint | int |
| real, double | float |
| numeric, decimal | decimal |
| bool | bool |
| char, varchar, text, clob | str |
| date | date |
| time, timetz | time |
| timestamp, timestamptz | datetime |
| interval | timedelta |
| bytea, blob | memoryview / bytes |
| array | list |
需要注意的地方：
- python 2里字符串是
str和unicode，python 3统一是str。 - 金仓的时间范围比python的
datetime大，比如date最大到9999-12-31，python也一样，这个没问题。 bytea类型在python 3里返回memoryview，需要用tobytes()转换才能拿到原始字节。
示例：
cur.execute("select bytea_column from test_lob")
row = cur.fetchone()
if isinstance(row[0], memoryview):
data = row[0].tobytes().decode('utf-8')
八、编码设置
连接时可以指定客户端编码，避免中文乱码：
conn = ksycopg2.connect(
database='test',
user='system',
password='123456',
client_encoding='utf8' # 指定编码
)
也可以在连接后动态设置：
conn.set_client_encoding('utf8')
如果不指定，默认用数据库的编码。建议统一用utf8。
九、小结
上篇主要讲了ksycopg2的安装配置、连接管理、高可用和连接池。总结几个要点：
ksycopg2是python连金仓的标准方式，支持python 2.7到3.13- 安装时注意依赖库
libkci的路径配置 - 高可用参数很重要：多主机、重试、自动找主、快速故障转移——这几个配好了，集群切换对应用影响小很多
- 连接池用
threadedconnectionpool，注意连接用完要putconn，不要close
以上就是python连接管理金仓数据库的完全指南的详细内容，更多关于python连接金仓数据库的资料请关注代码网其它相关文章！
发表评论