Python连接KingbaseES全指南-智慧文博士

好的，这是一个使用ksycopg2库在 Python 中连接和操作 KingbaseES 数据库的完整指南。

1. 安装依赖

首先，确保已安装ksycopg2包。它是psycopg2的一个分支，专为 KingbaseES 设计。

pip install ksycopg2

2. 建立数据库连接

导入库并使用连接参数创建连接对象。

import ksycopg2 # 配置连接参数 db_params = { "dbname": "your_database", # 数据库名 "user": "your_username", # 用户名 "password": "your_password", # 密码 "host": "localhost", # 主机地址 "port": "54321" # 端口（默认为54321） } try: # 建立连接 conn = ksycopg2.connect(**db_params) print("连接成功！") # 创建游标对象 cursor = conn.cursor() except ksycopg2.Error as e: print(f"连接失败: {e}")

3. 执行SQL查询

通过游标执行SQL语句，处理查询结果。

# 示例：查询数据 try: cursor.execute("SELECT id, name FROM users;") rows = cursor.fetchall() print("查询结果：") for row in rows: print(f"ID: {row[0]}, Name: {row[1]}") except ksycopg2.Error as e: print(f"查询失败: {e}")

4. 执行数据修改操作

插入、更新或删除数据时需显式提交事务。

# 示例：插入数据 try: cursor.execute( "INSERT INTO users (name, email) VALUES (%s, %s);", ("张三", "zhangsan@example.com") ) conn.commit() # 提交事务 print("数据插入成功！") except ksycopg2.Error as e: conn.rollback() # 回滚事务 print(f"插入失败: {e}")

5. 使用参数化查询

重要！使用参数化查询避免SQL注入。

user_id = 10 cursor.execute("SELECT * FROM logs WHERE user_id = %s;", (user_id,))

6. 执行DDL操作

创建表、修改结构等操作也需提交事务。

try: cursor.execute(""" CREATE TABLE products ( id SERIAL PRIMARY KEY, name VARCHAR(100) NOT NULL, price NUMERIC(10, 2) ); """) conn.commit() print("表创建成功！") except ksycopg2.Error as e: conn.rollback() print(f"DDL操作失败: {e}")

7. 错误处理

使用try-except捕获常见异常：

OperationalError: 连接问题
DataError: 数据格式错误
IntegrityError: 违反约束

try: cursor.execute("SELECT invalid_column FROM non_existent_table;") except ksycopg2.ProgrammingError as e: print(f"SQL语法错误: {e}")

8. 关闭连接

完成操作后释放资源。

# 关闭游标和连接 cursor.close() conn.close() print("连接已关闭")

完整示例代码

import ksycopg2 def main(): conn = None try: conn = ksycopg2.connect( dbname="test_db", user="admin", password="secure_pass", host="192.168.1.100", port="54321" ) cursor = conn.cursor() # 创建表 cursor.execute(""" CREATE TABLE IF NOT EXISTS employees ( id SERIAL PRIMARY KEY, name VARCHAR(50), department VARCHAR(50) ); """) # 插入数据 cursor.execute( "INSERT INTO employees (name, department) VALUES (%s, %s)", ("李四", "技术部") ) # 查询验证 cursor.execute("SELECT * FROM employees;") for record in cursor.fetchall(): print(record) conn.commit() except ksycopg2.DatabaseError as e: print(f"数据库错误: {e}") if conn: conn.rollback() finally: if conn: cursor.close() conn.close() if __name__ == "__main__": main()