Python连接KingbaseES数据库完整指南

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("数据库连接已关闭。")

# 假设游标 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)

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)

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)

# 写入大对象 (例如图片)
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)