pip install mysql-connector-python

-- 先創(chuàng)建測試表（可選）
CREATE TABLE IF NOT EXISTS users (
    id INT PRIMARY KEY AUTO_INCREMENT,
    name VARCHAR(50) NOT NULL,
    age INT,
    email VARCHAR(100)
);

-- 插入測試數(shù)據(jù)
INSERT INTO users (name, age, email) VALUES 
('張三', 25, 'zhangsan@test.com'),
('李四', 30, 'lisi@test.com');

-- 創(chuàng)建存儲過程：根據(jù)ID查詢用戶信息
DELIMITER //  -- 臨時修改語句結(jié)束符為//，避免與存儲過程內(nèi)的;沖突
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 ;  -- 恢復(fù)語句結(jié)束符為;

-- 另一個測試存儲過程：無參數(shù)，查詢所有用戶
CREATE PROCEDURE get_all_users()
BEGIN
    SELECT * FROM users;
END;

import mysql.connector
from mysql.connector import Error

def call_mysql_procedure():
    # 數(shù)據(jù)庫連接配置
    config = {
        'host': 'localhost',       # 數(shù)據(jù)庫地址
        'user': 'root',            # 用戶名
        'password': '你的數(shù)據(jù)庫密碼', # 密碼
        'database': 'test_db'      # 數(shù)據(jù)庫名（替換為你的庫名）
    }

    connection = None
    try:
        # 1. 建立數(shù)據(jù)庫連接
        connection = mysql.connector.connect(**config)
        if connection.is_connected():
            cursor = connection.cursor()

            # ========== 場景1：調(diào)用帶輸入/輸出參數(shù)的存儲過程 ==========
            # 定義輸入?yún)?shù)和輸出參數(shù)
            user_id = 1  # 輸入?yún)?shù)
            user_name = None  # 輸出參數(shù)（初始化為None）
            user_age = None   # 輸出參數(shù)（初始化為None）

            # 調(diào)用存儲過程：callproc(存儲過程名, (參數(shù)列表))
            cursor.callproc('get_user_by_id', (user_id, user_name, user_age))

            # 獲取輸出參數(shù)（存儲過程執(zhí)行后，參數(shù)會被更新，存在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：調(diào)用無參數(shù)的存儲過程（返回結(jié)果集） ==========
            print("場景2 - 查詢所有用戶：")
            cursor.callproc('get_all_users')  # 無參數(shù)，傳入空元組或省略
            # 遍歷結(jié)果集
            for result in cursor.stored_results():
                users = result.fetchall()  # 獲取所有結(jié)果
                # 打印表頭
                columns = [desc[0] for desc in result.description]
                print("\t".join(columns))
                # 打印數(shù)據(jù)
                for user in users:
                    print("\t".join(str(col) for col in user))

            # 提交事務(wù)（如果存儲過程有修改操作，必須提交）
            connection.commit()

    except Error as e:
        print(f"數(shù)據(jù)庫操作出錯：{e}")
        # 出錯時回滾事務(wù)
        if connection:
            connection.rollback()
    finally:
        # 關(guān)閉游標(biāo)和連接
        if connection and connection.is_connected():
            if cursor:
                cursor.close()
            connection.close()
            print("\n數(shù)據(jù)庫連接已關(guān)閉")

if __name__ == "__main__":
    call_mysql_procedure()