一、准备工作

首先需要安装 Python 操作 MySQL 的库，最常用的是 mysql-connector-python（官方库）或 pymysql，这里以官方库为例：

pip install mysql-connector-python

二、完整实现步骤

1. 先在 MySQL 中创建测试存储过程

首先我们需要一个可测试的存储过程，比如创建一个根据用户 ID 查询用户信息的存储过程：

-- 先创建测试表（可选）
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;

2. Python 调用存储过程的代码实现

下面是 Python 调用存储过程的完整代码，包含两种常见场景：带输入/输出参数的存储过程、无参数的存储过程：

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()