Python实现调用MySQL数据库的存储过程

pip install mysql-connector-python

-- 先创建测试表（可选）
CREATE TABLE IF NOT EXISTS users (
    id INT PRIMARY KEY AUTO_INCREMENT,
    name VARCHAR(50) NOT NULL,
    age INT,
    email VARCHAR(100)
);

-- 插入测试数据
INSERT INTO users (name, age, email) VALUES 
('张三', 25, 'zhangsan@test.com'),
('李四', 30, 'lisi@test.com');

-- 创建存储过程：根据ID查询用户信息
DELIMITER //  -- 临时修改语句结束符为//，避免与存储过程内的;冲突
CREATE PROCEDURE get_user_by_id(IN user_id INT, OUT user_name VARCHAR(50), OUT user_age INT)
BEGIN
    SELECT name, age INTO user_name, user_age FROM users WHERE id = user_id;
END //
DELIMITER ;  -- 恢复语句结束符为;

-- 另一个测试存储过程：无参数，查询所有用户
CREATE PROCEDURE get_all_users()
BEGIN
    SELECT * FROM users;
END;

import mysql.connector
from mysql.connector import Error

def call_mysql_procedure():
    # 数据库连接配置
    config = {
        'host': 'localhost',       # 数据库地址
        'user': 'root',            # 用户名
        'password': '你的数据库密码', # 密码
        'database': 'test_db'      # 数据库名（替换为你的库名）
    }

    connection = None
    try:
        # 1. 建立数据库连接
        connection = mysql.connector.connect(**config)
        if connection.is_connected():
            cursor = connection.cursor()

            # ========== 场景1：调用带输入/输出参数的存储过程 ==========
            # 定义输入参数和输出参数
            user_id = 1  # 输入参数
            user_name = None  # 输出参数（初始化为None）
            user_age = None   # 输出参数（初始化为None）

            # 调用存储过程：callproc(存储过程名, (参数列表))
            cursor.callproc('get_user_by_id', (user_id, user_name, user_age))

            # 获取输出参数（存储过程执行后，参数会被更新，存在cursor.stored_results()中）
            for result in cursor.stored_results():
                # 或者直接通过 cursor.outputs 获取（不同版本可能略有差异）
                output_params = result.fetchone()
                if output_params:
                    user_name, user_age = output_params
            print(f"场景1 - 查询ID为{user_id}的用户：")
            print(f"姓名：{user_name}，年龄：{user_age}\n")

            # ========== 场景2：调用无参数的存储过程（返回结果集） ==========
            print("场景2 - 查询所有用户：")
            cursor.callproc('get_all_users')  # 无参数，传入空元组或省略
            # 遍历结果集
            for result in cursor.stored_results():
                users = result.fetchall()  # 获取所有结果
                # 打印表头
                columns = [desc[0] for desc in result.description]
                print("\t".join(columns))
                # 打印数据
                for user in users:
                    print("\t".join(str(col) for col in user))

            # 提交事务（如果存储过程有修改操作，必须提交）
            connection.commit()

    except Error as e:
        print(f"数据库操作出错：{e}")
        # 出错时回滚事务
        if connection:
            connection.rollback()
    finally:
        # 关闭游标和连接
        if connection and connection.is_connected():
            if cursor:
                cursor.close()
            connection.close()
            print("\n数据库连接已关闭")

if __name__ == "__main__":
    call_mysql_procedure()