使用 Ksycopg2 驱动实现 Kingbase 数据库增删改查系统

本篇文章我将介绍如何使用Python的ksycopg2驱动连接Kingbase数据库，并使用Flask框架搭建一个简单的Web应用实现增删改查功能。

一、环境准备

本文使用的环境是，Python 3.12+，KingbaseV9R1C10，Ksycopg2驱动。在文章开始前，要确保电脑上面已经安装了开发环境。
Kingbase安装可参考以下的安装教程：
金仓数据库KingbaseES V9R1C10安装教程 - Windows版详细指南​
Kingbase官网：https://www.kingbase.com.cn/
Kingbase学习文档：https://docs.kingbase.com.cn/cn/KES-V9R1C10/introduction

验证Python版本：

Python -V

安装 Flask 框架和Ksycopg2 驱动：

# 安装 Flask 框架
pip3 install flask# 安装 Ksycopg2 驱动
pip3 install ksycopg2

Ksycopg2 是专为 Kingbase 数据库设计的 Python 驱动，其 API 设计与 PostgreSQL 的 Psycopg2 驱动高度兼容，这使得从 PostgreSQL 迁移到 Kingbase 变得非常容易。

二、Ksycopg2 基础使用

在开始构建完整的 Web 应用之前，我们先学习如何使用 Ksycopg2 驱动连接和操作 Kingbase 数据库。

2.1 测试数据库连接

创建一个测试文件 test_connection.py,验证是否能成功连接数据库：

重要提示：请将代码中的 user、password、host、port 和 database 替换为您自己的数据库配置信息。

# -*- 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))print("数据库连接成功!")# 获取数据库版本信息cur = conn.cursor()cur.execute("SELECT version()")version = cur.fetchone()print(f"数据库版本: {version[0]}")cur.close()conn.close()except Exception as e:print(f"数据库连接失败: {e}")

这里连接成功后咱们就可以继续进行下去了。此时已经说明我们使用ksycopg2成功驱动Kingbase

知识点：ksycopg2.connect() 方法使用连接字符串建立数据库连接。连接成功后，我们可以通过 cursor() 方法创建游标对象来执行 SQL 语句。

2.2 基础 CRUD 操作示例

创建 test_crud.py 文件，演示基本的增删改查操作：

# -*- coding: utf-8 -*-
import ksycopg2
from ksycopg2.extras import RealDictCursor# 数据库连接配置
DB_CONFIG = {'dbname': 'TEST','user': 'SYSTEM','password': '123456','host': '127.0.0.1','port': '54321'
}def get_connection():"""获取数据库连接"""return ksycopg2.connect("dbname={dbname} user={user} password={password} host={host} port={port}".format(**DB_CONFIG))# 1. 创建表 (Create Table)
def create_table():"""创建测试表"""conn = get_connection()cur = conn.cursor()try:# 删除已存在的表cur.execute('DROP TABLE IF EXISTS test_users')# 创建新表cur.execute('''CREATE TABLE test_users (id SERIAL PRIMARY KEY,username VARCHAR(50) NOT NULL,email VARCHAR(100),age INTEGER,created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP)''')conn.commit()print("表创建成功!")except Exception as e:print(f"创建表失败: {e}")conn.rollback()finally:cur.close()conn.close()# 2. 插入数据 (Create)
def insert_data():"""插入测试数据"""conn = get_connection()cur = conn.cursor()try:# 单条插入cur.execute("INSERT INTO test_users (username, email, age) VALUES (%s, %s, %s)",('张三', '[e-mail]', 25))# 批量插入users = [('李四', '[e-mail]', 28),('王五', '[e-mail]', 30),('赵六', '[e-mail]', 22),('测试用户', '[e-mail]', 35)]cur.executemany("INSERT INTO test_users (username, email, age) VALUES (%s, %s, %s)",users)conn.commit()print(f"成功插入 {len(users) + 1} 条数据!")except Exception as e:print(f"插入数据失败: {e}")conn.rollback()finally:cur.close()conn.close()# 3. 查询数据 (Read)
def query_data():"""查询数据"""conn = get_connection()# 使用 RealDictCursor 返回字典格式的结果cur = conn.cursor(cursor_factory=RealDictCursor)try:# 查询所有数据cur.execute("SELECT * FROM test_users ORDER BY id")users = cur.fetchall()print("\n所有用户列表:")print("-" * 80)for user in users:print(f"ID: {user['id']}, 姓名: {user['username']}, "f"邮箱: {user['email']}, 年龄: {user['age']}")print("-" * 80)# 条件查询cur.execute("SELECT * FROM test_users WHERE age > %s", (25,))filtered_users = cur.fetchall()print(f"\n年龄大于25的用户: {len(filtered_users)} 人")except Exception as e:print(f"查询数据失败: {e}")finally:cur.close()conn.close()# 4. 更新数据 (Update)
def update_data():"""更新数据"""conn = get_connection()cur = conn.cursor()try:# 更新指定用户的邮箱cur.execute("UPDATE test_users SET email = %s WHERE username = %s",('[e-mail]', '张三'))# 批量更新年龄cur.execute("UPDATE test_users SET age = age + 1 WHERE age < %s",(30,))conn.commit()print(f"更新了 {cur.rowcount} 条数据!")except Exception as e:print(f"更新数据失败: {e}")conn.rollback()finally:cur.close()conn.close()# 5. 删除数据 (Delete)
def delete_data():"""删除数据"""conn = get_connection()cur = conn.cursor()try:# 删除指定用户cur.execute("DELETE FROM test_users WHERE username = %s", ('测试用户',))conn.commit()print(f"删除了 {cur.rowcount} 条数据!")except Exception as e:print(f"删除数据失败: {e}")conn.rollback()finally:cur.close()conn.close()# 6. 统计查询
def statistics():"""统计查询"""conn = get_connection()cur = conn.cursor()try:# 总数统计cur.execute("SELECT COUNT(*) FROM test_users")total = cur.fetchone()[0]# 平均年龄cur.execute("SELECT AVG(age) FROM test_users")avg_age = cur.fetchone()[0]# 年龄范围cur.execute("SELECT MIN(age), MAX(age) FROM test_users")min_age, max_age = cur.fetchone()print(f"\n数据统计:")print(f"总用户数: {total}")print(f"平均年龄: {avg_age:.2f}")print(f"年龄范围: {min_age} - {max_age}")except Exception as e:print(f"统计查询失败: {e}")finally:cur.close()conn.close()# 主函数
if __name__ == '__main__':print("=" * 80)print("Ksycopg2 CRUD 操作演示")print("=" * 80)# 执行各项操作create_table()      # 创建表insert_data()       # 插入数据query_data()        # 查询数据update_data()       # 更新数据query_data()        # 再次查询查看更新结果delete_data()       # 删除数据query_data()        # 最终查询statistics()        # 统计信息print("\n所有操作完成!")

运行以下命令进行测试：

python test_connection.py

成功执行，创建表结构和CRUD操作没问题。

知识点总结：
execute() 用于执行单条 SQL 语句
executemany() 用于批量执行相同结构的 SQL 语句
fetchone() 获取单条结果
fetchall() 获取所有结果
commit() 提交事务
rollback() 回滚事务

2.3 Ksycopg2 关键特性说明

2.3.1 参数化查询

Ksycopg2 使用 %s 作为参数占位符，可以有效防止 SQL 注入：

# 正确的参数化查询
cur.execute("SELECT * FROM users WHERE id = %s", (user_id,))# 错误的字符串拼接(容易被 SQL 注入攻击)
cur.execute(f"SELECT * FROM users WHERE id = {user_id}")

安全提示：永远不要使用字符串拼接来构建 SQL 语句，这会导致严重的安全漏洞。参数化查询会自动处理特殊字符的转义，有效防止 SQL 注入攻击。

2.3.2 游标类型

# 默认游标 - 返回元组
cur = conn.cursor()
cur.execute("SELECT id, name FROM users")
result = cur.fetchone()  # (1, 'John')# 字典游标 - 返回字典(推荐)
from ksycopg2.extras import RealDictCursor
cur = conn.cursor(cursor_factory=RealDictCursor)
cur.execute("SELECT id, name FROM users")
result = cur.fetchone()  # {'id': 1, 'name': 'John'}

最佳实践：推荐使用 RealDictCursor，因为字典格式的结果更易读，代码可维护性更好。

2.3.3 事务管理

conn = get_connection()
cur = conn.cursor()try:# 开始事务(自动开始)cur.execute("INSERT INTO users ...")cur.execute("UPDATE accounts ...")# 提交事务conn.commit()
except Exception as e:# 回滚事务conn.rollback()print(f"事务失败: {e}")
finally:cur.close()conn.close()

事务特性：事务保证了数据操作的原子性（要么全部成功，要么全部失败），这对于保持数据一致性至关重要。

三、完整 Web 应用实战

通过前面的学习，我们已经掌握了 Ksycopg2 的基础用法。现在让我们将这些知识应用到实际项目中，构建一个功能完整的学生管理系统。

3.1 项目结构

看一下整体的项目结构

student_system/
├── app.py              # 主程序
├── database.py         # 数据库操作
└── templates/├── index.html      # 列表页面├── add.html        # 添加页面└── edit.html       # 编辑页面

设计思想：采用 MVC 架构模式，将数据库操作（Model）、业务逻辑（Controller）和页面展示（View）分离，提高代码的可维护性和可扩展性。

3.2 database.py (数据库操作)

# -*- coding: utf-8 -*-
import ksycopg2
from ksycopg2.extras import RealDictCursorclass Database:def __init__(self):self.config = {'dbname': 'TEST','user': 'SYSTEM','password': '123456','host': '127.0.0.1','port': '54321'}def get_connection(self):"""获取数据库连接"""try:conn = ksycopg2.connect("dbname={dbname} user={user} password={password} host={host} port={port}".format(**self.config))return connexcept Exception as e:print(f"数据库连接失败: {e}")return Nonedef init_table(self):"""初始化数据表"""conn = self.get_connection()if not conn:return Falsetry:cur = conn.cursor()cur.execute('''CREATE TABLE IF NOT EXISTS students (id SERIAL PRIMARY KEY,student_no VARCHAR(50) NOT NULL,name VARCHAR(100) NOT NULL,age INTEGER,gender VARCHAR(10),major VARCHAR(100),created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP)''')conn.commit()cur.close()conn.close()print("数据表初始化成功!")return Trueexcept Exception as e:print(f"初始化表失败: {e}")return Falsedef add_student(self, student_no, name, age, gender, major):"""添加学生"""conn = self.get_connection()if not conn:return Falsetry:cur = conn.cursor()cur.execute("INSERT INTO students (student_no, name, age, gender, major) VALUES (%s, %s, %s, %s, %s)",(student_no, name, age, gender, major))conn.commit()cur.close()conn.close()return Trueexcept Exception as e:print(f"添加学生失败: {e}")return Falsedef get_all_students(self):"""查询所有学生"""conn = self.get_connection()if not conn:return []try:cur = conn.cursor(cursor_factory=RealDictCursor)cur.execute("SELECT * FROM students ORDER BY id DESC")students = cur.fetchall()cur.close()conn.close()return studentsexcept Exception as e:print(f"查询学生失败: {e}")return []def get_student_by_id(self, student_id):"""根据ID查询学生"""conn = self.get_connection()if not conn:return Nonetry:cur = conn.cursor(cursor_factory=RealDictCursor)cur.execute("SELECT * FROM students WHERE id = %s", (student_id,))student = cur.fetchone()cur.close()conn.close()return studentexcept Exception as e:print(f"查询学生失败: {e}")return Nonedef update_student(self, student_id, student_no, name, age, gender, major):"""更新学生信息"""conn = self.get_connection()if not conn:return Falsetry:cur = conn.cursor()cur.execute("UPDATE students SET student_no=%s, name=%s, age=%s, gender=%s, major=%s WHERE id=%s",(student_no, name, age, gender, major, student_id))conn.commit()cur.close()conn.close()return Trueexcept Exception as e:print(f"更新学生失败: {e}")return Falsedef delete_student(self, student_id):"""删除学生"""conn = self.get_connection()if not conn:return Falsetry:cur = conn.cursor()cur.execute("DELETE FROM students WHERE id = %s", (student_id,))conn.commit()cur.close()conn.close()return Trueexcept Exception as e:print(f"删除学生失败: {e}")return False

设计亮点：
使用类封装数据库操作，提高代码复用性
统一的异常处理机制
自动管理数据库连接的打开和关闭
所有方法都有明确的返回值，便于调用者判断操作是否成功

3.3 app.py (Flask应用)

# -*- coding: utf-8 -*-
from flask import Flask, render_template, request, redirect, url_for, flash
from database import Databaseapp = Flask(__name__)
app.secret_key = 'your_secret_key_here'# 初始化数据库
db = Database()
db.init_table()@app.route('/')
def index():"""主页 - 显示所有学生"""students = db.get_all_students()return render_template('index.html', students=students)@app.route('/add', methods=['GET', 'POST'])
def add():"""添加学生"""if request.method == 'POST':student_no = request.form.get('student_no')name = request.form.get('name')age = request.form.get('age')gender = request.form.get('gender')major = request.form.get('major')if db.add_student(student_no, name, age, gender, major):flash('学生添加成功!', 'success')return redirect(url_for('index'))else:flash('学生添加失败!', 'error')return render_template('add.html')@app.route('/edit/<int:student_id>', methods=['GET', 'POST'])
def edit(student_id):"""编辑学生"""if request.method == 'POST':student_no = request.form.get('student_no')name = request.form.get('name')age = request.form.get('age')gender = request.form.get('gender')major = request.form.get('major')if db.update_student(student_id, student_no, name, age, gender, major):flash('学生信息更新成功!', 'success')return redirect(url_for('index'))else:flash('学生信息更新失败!', 'error')student = db.get_student_by_id(student_id)return render_template('edit.html', student=student)@app.route('/delete/<int:student_id>')
def delete(student_id):"""删除学生"""if db.delete_student(student_id):flash('学生删除成功!', 'success')else:flash('学生删除失败!', 'error')return redirect(url_for('index'))if __name__ == '__main__':app.run(debug=True, host='0.0.0.0', port=5000)

Flask 路由说明：
@app.route(‘/’) - 主页，显示学生列表
@app.route(‘/add’) - 添加学生（支持 GET 和 POST）
@app.route(‘/edit/int:student_id’) - 编辑学生（动态路由参数）
@app.route(‘/delete/int:student_id’) - 删除学生
flash() 函数用于显示操作提示消息

3.4 templates/index.html (列表页面)

<!DOCTYPE html>
<html lang="zh-CN">
<head><meta charset="UTF-8"><meta name="viewport" content="width=device-width, initial-scale=1.0"><title>学生管理系统</title><style>* { margin: 0; padding: 0; box-sizing: border-box; }body { font-family: Arial, 'Microsoft YaHei', sans-serif; padding: 20px; background: #f5f5f5; }.container { max-width: 1200px; margin: 0 auto; background: white; padding: 20px; border-radius: 5px; }h1 { color: #333; margin-bottom: 20px; }.alert { padding: 10px; margin-bottom: 15px; border-radius: 3px; }.alert-success { background: #d4edda; color: #155724; border: 1px solid #c3e6cb; }.alert-error { background: #f8d7da; color: #721c24; border: 1px solid #f5c6cb; }.btn { display: inline-block; padding: 8px 15px; text-decoration: none; border-radius: 3px; margin-bottom: 15px; }.btn-primary { background: #007bff; color: white; }.btn-success { background: #28a745; color: white; }.btn-warning { background: #ffc107; color: #333; }.btn-danger { background: #dc3545; color: white; }table { width: 100%; border-collapse: collapse; margin-top: 10px; }th, td { padding: 12px; text-align: left; border: 1px solid #ddd; }th { background: #007bff; color: white; }tr:hover { background: #f5f5f5; }.no-data { text-align: center; padding: 40px; color: #999; }</style>
</head>
<body><div class="container"><h1>🎓 学生管理系统</h1>{% with messages = get_flashed_messages(with_categories=true) %}{% if messages %}{% for category, message in messages %}<div class="alert alert-{{ category }}">{{ message }}</div>{% endfor %}{% endif %}{% endwith %}<a href="{{ url_for('add') }}" class="btn btn-primary">➕ 添加学生</a>{% if students %}<table><thead><tr><th>ID</th><th>学号</th><th>姓名</th><th>年龄</th><th>性别</th><th>专业</th><th>创建时间</th><th>操作</th></tr></thead><tbody>{% for student in students %}<tr><td>{{ student.id }}</td><td>{{ student.student_no }}</td><td>{{ student.name }}</td><td>{{ student.age }}</td><td>{{ student.gender }}</td><td>{{ student.major }}</td><td>{{ student.created_at }}</td><td><a href="{{ url_for('edit', student_id=student.id) }}" class="btn btn-warning">编辑</a><a href="{{ url_for('delete', student_id=student.id) }}" class="btn btn-danger" onclick="return confirm('确定要删除吗?')">删除</a></td></tr>{% endfor %}</tbody></table>{% else %}<div class="no-data">暂无学生数据</div>{% endif %}</div>
</body>
</html>

3.5 templates/add.html (添加页面)

<!DOCTYPE html>
<html lang="zh-CN">
<head><meta charset="UTF-8"><meta name="viewport" content="width=device-width, initial-scale=1.0"><title>添加学生</title><style>* { margin: 0; padding: 0; box-sizing: border-box; }body { font-family: Arial, 'Microsoft YaHei', sans-serif; padding: 20px; background: #f5f5f5; }.container { max-width: 600px; margin: 0 auto; background: white; padding: 20px; border-radius: 5px; }h1 { color: #333; margin-bottom: 20px; }.form-group { margin-bottom: 15px; }label { display: block; margin-bottom: 5px; font-weight: bold; }input, select { width: 100%; padding: 8px; border: 1px solid #ddd; border-radius: 3px; }.btn { display: inline-block; padding: 10px 20px; text-decoration: none; border: none; border-radius: 3px; cursor: pointer; margin-right: 10px; }.btn-success { background: #28a745; color: white; }.btn-secondary { background: #6c757d; color: white; }</style>
</head>
<body><div class="container"><h1>➕ 添加学生</h1><form method="POST"><div class="form-group"><label>学号:</label><input type="text" name="student_no" required></div><div class="form-group"><label>姓名:</label><input type="text" name="name" required></div><div class="form-group"><label>年龄:</label><input type="number" name="age" min="1" max="150" required></div><div class="form-group"><label>性别:</label><select name="gender" required><option value="">请选择</option><option value="男">男</option><option value="女">女</option></select></div><div class="form-group"><label>专业:</label><input type="text" name="major" required></div><button type="submit" class="btn btn-success">保存</button><a href="{{ url_for('index') }}" class="btn btn-secondary">返回</a></form></div>
</body>
</html>

3.6 templates/edit.html (编辑页面)

<!DOCTYPE html>
<html lang="zh-CN">
<head><meta charset="UTF-8"><meta name="viewport" content="width=device-width, initial-scale=1.0"><title>编辑学生</title><style>* { margin: 0; padding: 0; box-sizing: border-box; }body { font-family: Arial, 'Microsoft YaHei', sans-serif; padding: 20px; background: #f5f5f5; }.container { max-width: 600px; margin: 0 auto; background: white; padding: 20px; border-radius: 5px; }h1 { color: #333; margin-bottom: 20px; }.form-group { margin-bottom: 15px; }label { display: block; margin-bottom: 5px; font-weight: bold; }input, select { width: 100%; padding: 8px; border: 1px solid #ddd; border-radius: 3px; }.btn { display: inline-block; padding: 10px 20px; text-decoration: none; border: none; border-radius: 3px; cursor: pointer; margin-right: 10px; }.btn-success { background: #28a745; color: white; }.btn-secondary { background: #6c757d; color: white; }</style>
</head>
<body><div class="container"><h1>✏️ 编辑学生</h1>{% if student %}<form method="POST"><div class="form-group"><label>学号:</label><input type="text" name="student_no" value="{{ student.student_no }}" required></div><div class="form-group"><label>姓名:</label><input type="text" name="name" value="{{ student.name }}" required></div><div class="form-group"><label>年龄:</label><input type="number" name="age" value="{{ student.age }}" min="1" max="150" required></div><div class="form-group"><label>性别:</label><select name="gender" required><option value="男" {% if student.gender == '男' %}selected{% endif %}>男</option><option value="女" {% if student.gender == '女' %}selected{% endif %}>女</option></select></div><div class="form-group"><label>专业:</label><input type="text" name="major" value="{{ student.major }}" required></div><button type="submit" class="btn btn-success">更新</button><a href="{{ url_for('index') }}" class="btn btn-secondary">返回</a></form>{% else %}<p>学生不存在</p><a href="{{ url_for('index') }}" class="btn btn-secondary">返回</a>{% endif %}</div>
</body>
</html>

3.5 运行和测试应用

3.5.1 运行应用

在项目根目录下执行：

python app.py

访问系统：打开浏览器访问: http://localhost:5000

3.5.2 测试添加功能

填写学生信息后点击保存：

3.5.3 测试编辑功能

点击编辑按钮：

修改信息后保存：

3.5.4 测试删除功能

点击删除按钮，确认后删除：

四、文章总结

本文详细介绍了如何使用 Python 的 Ksycopg2 驱动连接 Kingbase 数据库，并基于 Flask 框架构建了一个完整的学生管理系统。文章从环境准备开始，循序渐进地讲解了 Ksycopg2 的基础使用方法，包括数据库连接、CRUD 操作、参数化查询、事务管理等核心知识点。随后通过实战项目演示了如何将数据库操作封装成独立的 Database 类，采用 MVC 架构模式实现了学生信息的增删改查功能，并提供了完整的前端页面展示。整个教程代码完整、注释详细，适合初学者快速掌握 Python + Kingbase 的 Web 应用开发技能。

---

关键技术栈总结：

• 数据库：Kingbase V9R1C10
• Python 版本：3.12+
• 核心驱动：Ksycopg2（兼容 Psycopg2 API）
• Web 框架：Flask
• 架构模式：MVC（Model-View-Controller）
• 安全特性：参数化查询防止 SQL 注入、事务管理保证数据一致性