Python连接KingbaseES数据库完整指南_Python

这是一份使用 psycopg2 连接 kingbasees 数据库的完整指南。kingbasees 与 postgresql 高度兼容，因此 psycopg2（postgresql 的 python 适配器）是连接它的常用工具。

1. 环境准备

安装 psycopg2：在命令行中执行：
```
pip install psycopg2-binary
```
psycopg2-binary 包含了预编译的二进制文件，安装更方便。如果需要从源码编译，可以安装 psycopg2。
确认 kingbasees 服务：确保你的 kingbasees 数据库服务已启动，并且你拥有目标数据库的连接权限（用户名、密码、数据库名、主机地址、端口号）。

2. 建立基本连接

import psycopg2
# 连接参数
dbname = "your_database_name"  # 数据库名
user = "your_username"        # 用户名
password = "your_password"    # 密码
host = "your_host_address"    # 数据库服务器地址，如 'localhost' 或 ip
port = "your_port"            # 端口号，默认通常是 54321 或 5432 (取决于kingbasees配置)
try:
    # 建立连接
    connection = psycopg2.connect(
        dbname=dbname,
        user=user,
        password=password,
        host=host,
        port=port
    )
    print("成功连接到 kingbasees 数据库！")
    # 创建一个游标对象来执行 sql 语句
    cursor = connection.cursor()
    # ... 后续操作 (查询、插入等) ...
except (exception, psycopg2.error) as error:
    print("连接或操作过程中发生错误:", error)
finally:
    # 确保在结束时关闭连接和游标
    if connection:
        cursor.close()
        connection.close()
        print("数据库连接已关闭。")

3. 执行查询 (select)

# 假设游标 cursor 已经创建
try:
    # 执行 sql 查询语句
    cursor.execute("select version();")  # 查询数据库版本信息
    # 获取结果
    db_version = cursor.fetchone()  # 获取单行结果
    print("kingbasees 版本信息:", db_version)
    # 查询多行数据示例
    cursor.execute("select id, name from your_table where status = %s;", ('active',))
    records = cursor.fetchall()  # 获取所有结果行
    print("查询结果:")
    for row in records:
        print(f"id: {row[0]}, name: {row[1]}")
except (exception, psycopg2.error) as error:
    print("查询执行错误:", error)

注意：

使用 cursor.execute(sql_query) 执行 sql 语句。
参数化查询使用 %s 作为占位符，并将参数作为元组传入 execute() 的第二个参数。这能有效防止 sql 注入攻击。
fetchone() 获取下一行。
fetchall() 获取所有剩余行。
fetchmany(size=n) 获取指定数量的行。

4. 执行数据操作 (insert, update, delete)

try:
    # 插入数据
    insert_query = """insert into your_table (id, name, email)
                      values (%s, %s, %s);"""
    record_to_insert = (101, '张三', 'zhangsan@example.com')
    cursor.execute(insert_query, record_to_insert)
    # 更新数据
    update_query = """update your_table set name = %s where id = %s;"""
    cursor.execute(update_query, ("李四", 101))
    # 删除数据
    delete_query = """delete from your_table where id = %s;"""
    cursor.execute(delete_query, (102,))
    # 提交事务！非常重要，否则更改不会持久化到数据库
    connection.commit()
    print("事务已提交。")
except (exception, psycopg2.error) as error:
    # 如果发生错误，回滚事务
    if connection:
        connection.rollback()
    print("操作执行错误:", error)

关键点：

数据修改操作（insert/update/delete）后，必须调用 connection.commit() 来提交事务，使更改永久生效。
如果操作过程中出错，调用 connection.rollback() 可以撤销当前事务中的所有更改。

5. 使用连接池 (可选，推荐用于高并发)

频繁创建和关闭数据库连接开销较大。使用连接池可以提高性能。

from psycopg2 import pool
# 创建连接池 (这里使用简单的 threadedconnectionpool)
min_conn = 1
max_conn = 5
connection_pool = pool.threadedconnectionpool(min_conn, max_conn,
                                             dbname=dbname,
                                             user=user,
                                             password=password,
                                             host=host,
                                             port=port)
try:
    # 从连接池获取一个连接
    connection = connection_pool.getconn()
    cursor = connection.cursor()
    cursor.execute("select ...")
    # ... 执行操作 ...
    connection.commit()
except (exception, psycopg2.error) as error:
    print("错误:", error)
finally:
    # 将连接放回连接池
    if connection:
        cursor.close()
        connection_pool.putconn(connection)

6. 处理大对象 (blob/clob)

kingbasees 使用 oid 和 lo 函数处理大对象。

# 写入大对象 (例如图片)
with open('image.jpg', 'rb') as f:
    data = f.read()
    # 创建大对象并获取其 oid
    cursor.execute("insert into your_blob_table (file_name) values (%s) returning file_oid;", ('image.jpg',))
    file_oid = cursor.fetchone()[0]
    # 打开大对象进行写入
    lobject = connection.lobject(file_oid, 'wb')  # 'wb' 表示写入二进制模式
    lobject.write(data)
    lobject.close()
    connection.commit()
# 读取大对象
cursor.execute("select file_oid from your_blob_table where file_name = %s;", ('image.jpg',))
file_oid = cursor.fetchone()[0]
lobject = connection.lobject(file_oid, 'rb')  # 'rb' 表示读取二进制模式
data = lobject.read()
lobject.close()
with open('retrieved_image.jpg', 'wb') as f:
    f.write(data)

7. 常见问题排查

连接失败：
- 检查 host, port, dbname, user, password 是否正确。
- 确认数据库服务是否正在运行 (netstat -an | grep <port> 或查看服务状态)。
- 检查防火墙是否允许连接。
- 查看 kingbasees 的日志文件（通常位于数据库安装目录的 data 子目录下）获取错误详情。
尝试使用 sslmode='require' 等参数（如果需要 ssl）。
psycopg2.operationalerror：通常表示网络或连接配置问题。
psycopg2.programmingerror：通常表示 sql 语法错误或对象不存在。
忘记 commit()：数据修改操作后必须提交，否则更改会丢失。
字符编码问题：确保数据库和客户端的编码一致（如 utf-8）。