示例

环境说明

Python 3.8.0
dmPython                  2.5.8

代码

# -*- coding: UTF-8 -*-
import sys

import dmPython
import os

# -------------------------- 数据库连接配置 --------------------------
DB_CONFIG = {
    "server": "127.0.0.1",       # 数据库主机
    "port": 5237,              # 端口
    "user": "用户名",            # 用户名
    "password": "密码",  # 数据库密码（替换为自己的）
    "cursorclass":dmPython.DictCursor # 返回类型
}

def get_db_connection():
    """获取数据库连接（封装为函数，方便复用）"""
    try:
        conn=dmPython.connect(**DB_CONFIG)
        return conn
    except (dmPython.Error, Exception) as e:
        print(f"数据库连接失败：{e}")
        return None

def init_table():
    """初始化测试表（用户表），仅执行一次"""
    conn = get_db_connection()
    if not conn:
        return
    try:
        cursor = conn.cursor()
        # 创建用户表（id自增主键，name+age两个字段）
        create_sql = """
        CREATE TABLE IF NOT EXISTS users (
            id INT IDENTITY(1,1) PRIMARY KEY,
            name VARCHAR(50) NOT NULL,
            age INT NOT NULL
        ) ;
        """
        cursor.execute(create_sql)
        conn.commit()
        print("表初始化成功（若不存在则创建）")
    except (dmPython.Error, Exception) as e:
        print(f"建表失败：{e}")
        conn.rollback()
    finally:
        cursor.close()
        conn.close()

# -------------------------- 预编译 CRUD 操作 --------------------------
def add_user(name: str, age: int):
    """新增用户（预编译 INSERT）"""
    conn = get_db_connection()
    if not conn:
        return False
    try:
        cursor = conn.cursor()
        # 预编译 SQL：用 %s 作为占位符（pymysql 专用，不是字符串格式化）
        insert_sql = "INSERT INTO users (name, age) VALUES (:1, :2);"
        # 执行预编译：参数以元组形式传递，自动转义防注入
        cursor.execute(insert_sql, (name, age))
        conn.commit()  # 增删改必须提交事务
        print(f"新增用户成功：{name}（ID：{cursor.lastrowid}）")
        return True
    except (dmPython.Error, Exception) as e:
        print(f"新增用户失败：{e}")
        conn.rollback()  # 失败回滚事务
        return False
    finally:
        cursor.close()
        conn.close()

def delete_user(user_id: int):
    """删除用户（预编译 DELETE）"""
    conn = get_db_connection()
    if not conn:
        return False
    try:
        cursor = conn.cursor()
        delete_sql = "DELETE FROM users WHERE id = :1 ;"
        cursor.execute(delete_sql, (user_id,))  # 单参数需加逗号，保持元组格式
        conn.commit()
        if cursor.rowcount > 0:
            print(f"删除用户成功：ID={user_id}")
            return True
        else:
            print(f"删除失败：未找到 ID={user_id} 的用户")
            return False
    except (dmPython.Error, Exception) as e:
        print(f"删除用户失败：{e}")
        conn.rollback()
        return False
    finally:
        cursor.close()
        conn.close()

def update_user(user_id: int, new_age: int):
    """更新用户年龄（预编译 UPDATE）"""
    conn = get_db_connection()
    if not conn:
        return False
    try:
        cursor = conn.cursor()
        update_sql = "UPDATE users SET age = :1 WHERE id = :2 ;"
        cursor.execute(update_sql, (new_age, user_id))
        conn.commit()
        if cursor.rowcount > 0:
            print(f"更新用户成功：ID={user_id}，新年龄={new_age}")
            return True
        else:
            print(f"更新失败：未找到 ID={user_id} 的用户")
            return False
    except (dmPython.Error, Exception) as e:
        print(f"更新用户失败：{e}")
        conn.rollback()
        return False
    finally:
        cursor.close()
        conn.close()

def query_users(age: int = None):
    """查询用户（预编译 SELECT）：支持按年龄筛选，无参数则查全部"""
    conn = get_db_connection()
    if not conn:
        return []
    try:
        # cursor = conn.cursor(dmPython.cursor)  # 返回字典格式结果
        cursor = conn.cursor()  # 返回字典格式结果
        if age:
            # 带参数的预编译查询
            query_sql = "SELECT id, name, age FROM users WHERE age = :1;"
            cursor.execute(query_sql, (age,))
        else:
            # 无参数查询（仍用预编译语法，仅无参数）
            query_sql = "SELECT id, name, age FROM users;"
            cursor.execute(query_sql)
        # 获取查询结果
        results = cursor.fetchall()
        print(f"查询到 {len(results)} 条用户数据：")
        for user in results:
            print(f"ID:{user['ID']} | 姓名:{user['NAME']} | 年龄:{user['AGE']}")
        return results
    except (dmPython.Error, Exception) as e:
        print(f"查询用户失败：{e}")
        return []
    finally:
        cursor.close()
        conn.close()

# -------------------------- 测试执行 --------------------------
if __name__ == "__main__":
    # 1. 初始化表（首次运行执行）
    init_table()

    # 2. 新增用户（预编译 INSERT）
    add_user("张三", 25)
    add_user("李四", 30)
    add_user("王五", 25)  # 年龄重复，测试查询筛选

    # 3. 查询全部用户
    query_users()

    # 4. 按年龄筛选查询
    query_users(age=25)

    # 5. 更新用户
    update_user(user_id=1, new_age=26)

    # 6. 删除用户
    delete_user(user_id=3)

    # 7. 再次查询验证结果
    query_users()
    sys.exit(0)

执行效果

表初始化成功（若不存在则创建）
新增用户成功：张三（ID：AAAAAAAAAAAAAAAAAB）
新增用户成功：李四（ID：AAAAAAAAAAAAAAAAAC）
新增用户成功：王五（ID：AAAAAAAAAAAAAAAAAD）
查询到 3 条用户数据：
ID:1 | 姓名:张三 | 年龄:25
ID:2 | 姓名:李四 | 年龄:30
ID:3 | 姓名:王五 | 年龄:25
查询到 2 条用户数据：
ID:1 | 姓名:张三 | 年龄:25
ID:3 | 姓名:王五 | 年龄:25
更新用户成功：ID=1，新年龄=26
删除用户成功：ID=3
查询到 2 条用户数据：
ID:1 | 姓名:张三 | 年龄:26
ID:2 | 姓名:李四 | 年龄:30

注意事项

1. 大小写敏感

当配置 “cursorclass”:dmPython.DictCursor 时，需要注意 数据库大小写敏感，确保返回字典中列名与获取列中保持一致