使用Python+XAMPP开发折扣卡系统时遇MySQL Connection not available错误

问题背景 #

开发集成Stripe支付与用户注册功能的折扣卡系统，采用MySQL+XAMPP搭建数据库。此前系统运行正常，提交至GitHub私有仓库后开发折扣卡模块时出现「MySQL Connection not available」错误，回滚至备份版本后问题仍存在。

现有代码 #

from flask import Flask, render_template, request, redirect, url_for, flash
import datetime
from flask import session
import mysql.connector
import stripe

stripe.api_key = "api_key"

app = Flask(__name__)
app.secret_key = 'secret_key'
app.config['SESSION_PERMANENT'] = False
app.config['SESSION_TYPE'] = 'filesystem'
app.config['PERMANENT_SESSION_LIFETIME'] = datetime.timedelta(days=10)

try:
    db_connection = mysql.connector.connect(
        host="host",
        user="user",
        password="password",
        database="database"
    )
    print("Conectado")
except mysql.connector.Error as e:
    print("Conexão falhou:", e)
db_connection.close()


def email_exists(email):
    cursor = db_connection.cursor()
    verificar_query = "SELECT id FROM usuarios WHERE email = %s"
    cursor.execute(verificar_query, (email,))
    existe = cursor.fetchone()
    db_connection.close()

    return existe is not None


@app.route('/form', methods=['GET', 'POST'])
def index():
    if request.method == 'POST':
        email = request.form.get('email')  # Obtém o email do formulário

        if email_exists(email):
            flash("O email já está em uso. Não é possível cadastrar.")
    else:
        if request.method == 'POST':
            name = request.form.get('name')
            email = request.form.get('email')
            password = request.form.get('password')
            nascimento = request.form.get('dataNascimento')
            telefone = request.form.get('tel')
            rg = request.form.get('rg')
            cpf = request.form.get('cpf')
            endereco = request.form.get('adress')
            cidade = request.form.get('city')
            numero = request.form.get('num')
            uf = request.form.get('top')
            cep = request.form.get('cep')
            complemento = request.form.get('com')

            cursor = db_connection.cursor()

            sql = "INSERT INTO usuarios (name, email, password, nascimento, endereco, cidade, cpf, telefone, numero, uf, rg, cep, complemento) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)"
            valores = (name, email, password, nascimento, telefone, rg, cpf, endereco, cidade, numero, uf, cep, complemento)
            app.config['PERMANENT_SESSION_LIFETIME'] = datetime.timedelta(days=10)
            cursor.execute(sql, valores)
            db_connection.commit()
            cursor.close()

    return render_template('registro-associado.html')


@app.route('/', methods=['GET', 'POST'])
def html():
    return render_template('index.html')


def login(name, email, password, cpf):
    cursor = db_connection.cursor()

    query = "SELECT * FROM usuarios WHERE name = %s AND email = %s AND password = %s AND cpf = %s"
    cursor.execute(query, (name, email, password, cpf))
        
    # user = cursor.fetchone()
    # sql_query = """
    # UPDATE usuarios SET numero_cartao = CONCAT(FLOOR(1000 + (RAND() * 9000)), ' ', FLOOR(1000 + (RAND() * 9000)), ' ', FLOOR(1000 + (RAND() * 9000)), ' ', FLOOR(1000 + (RAND() * 9000)));
    # """
    # cursor.execute(sql_query,)

    cursor.close()
    
    db_connection.close()

    return user


@app.route('/login', methods=['GET', 'POST'])
def login_page():
    if request.method == 'POST':
        name = request.form['name']
        email = request.form['email']
        password = request.form['password']
        cpf = request.form['cpf']

        user = login(name, email, password, cpf)

        if user:
            session['name'] = user[1]
            session['email'] = user[2]
            session['password'] = user[3]
            session['cpf'] = user[7]
            # session['numero_cartao'] = user[15]
            flash(f"Login bem-sucedido! Bem-vindo, {user[1]}", 'success')
            return redirect(url_for('cartao'))
        else:
            flash("Credenciais inválidas. Tente novamente.", 'error')

    return render_template('login.html')


def make_payment(client_secret):
    try:
        payment_intent = stripe.PaymentIntent.confirm(client_secret)
        if payment_intent.status == "succeeded":
            name = request.form['name']
            email = request.form['email']
            password = request.form['password']
            cpf = request.form['cpf']
            user = login(name, email, password, cpf)
            cursor = db_connection.cursor()
            # query = "UPDATE usuarios SET cartao = 1 WHERE id = %s"
            # cursor.execute(query, (user[0],))
            # sql_query = "UPDATE usuarios SET numero_cartao = CONCAT(FLOOR(1000 + (RAND() * 9000)), ' ', FLOOR(1000 + (RAND() * 9000)), ' ', FLOOR(1000 + (RAND() * 9000)), ' ', FLOOR(1000 + (RAND() * 9000)));"
            # cursor.execute(sql_query, (user[0],))
            db_connection.commit()
            db_connection.close()
        if payment_intent.status == "requires_payment_method":
            name = request.form['name']
            email = request.form['email']
            password = request.form['password']
            cpf = request.form['cpf']
            user = login(name, email, password, cpf)
            cursor = db_connection.cursor()
            # query = "UPDATE usuarios SET cartao = 0 WHERE id = %s"
            # cursor.execute(query, (user[0],))
            db_connection.commit()
            db_connection.close()
    except stripe.error.StripeError as e:
        print(e)


client_secret = "client_secret"
make_payment(client_secret)


@app.route('/logout')
def logout():
    session.clear()
    flash("Logout realizado com sucesso.", 'success')
    return redirect(url_for('index'))


@app.route('/index')
def page():
    if 'name' in session:
        return render_template('index.html', name=session['name'], email=session['email'], password=session['password'], cpf=session['cpf'])
        # return render_template('index.html', name=session['name'], email=session['email'], password=session['password'], cpf=session['cpf'], number=session['numero_cartao'])
    else:
        return "Acesso negado. Faça o login primeiro."
    

@app.route('/cartao')
def cart():
    if 'name' in session:
        return render_template('cartao.html', name=session['name'], email=session['email'], password=session['password'], cpf=session['cpf'])
        # return render_template('index.html', name=session['name'], email=session['email'], password=session['password'], cpf=session['cpf'], number=session['numero_cartao'])
    else:
        return "Acesso negado. Faça o login primeiro."
    

@app.route('/cartao')
def cartao():
    return render_template('cartao.html')


if __name__ == '__main__':
    app.run(debug=True)

错误原因及修复方案 #

核心问题：数据库连接被提前关闭 #

代码启动时创建连接后立刻执行db_connection.close()，后续所有数据库操作都在调用已关闭的连接，直接导致「MySQL Connection not available」错误。

其他关键问题 #

1. 函数内重复关闭连接：email_exists、login、make_payment等函数多次关闭全局连接，即使修复初始关闭问题，后续操作仍会失效。
2. 路由逻辑错误：
   • /form路由中else分支嵌套if request.method == 'POST'，该分支永远无法触发，用户注册逻辑完全失效。
   • /cartao路由重复定义，第二个cartao函数会覆盖第一个带权限验证的版本。
3. 变量未定义：login函数中user变量被注释，返回时会抛出未定义错误。
4. 启动时非法调用make_payment：启动时直接调用该函数，此时request对象不存在，会引发错误。
5. 安全隐患：明文存储用户密码，存在泄露风险。

修复后的代码示例 #

from flask import Flask, render_template, request, redirect, url_for, flash
import datetime
from flask import session
import mysql.connector
from mysql.connector import Error
import stripe
from werkzeug.security import generate_password_hash, check_password_hash

stripe.api_key = "api_key"

app = Flask(__name__)
app.secret_key = 'secret_key'
app.config['SESSION_PERMANENT'] = False
app.config['SESSION_TYPE'] = 'filesystem'
app.config['PERMANENT_SESSION_LIFETIME'] = datetime.timedelta(days=10)

# 数据库连接辅助函数，每次操作获取新连接
def get_db_connection():
    try:
        conn = mysql.connector.connect(
            host="host",
            user="user",
            password="password",
            database="database"
        )
        return conn
    except Error as e:
        print(f"数据库连接错误: {e}")
        return None


def email_exists(email):
    conn = get_db_connection()
    if not conn:
        return False
    cursor = conn.cursor()
    verificar_query = "SELECT id FROM usuarios WHERE email = %s"
    cursor.execute(verificar_query, (email,))
    existe = cursor.fetchone()
    cursor.close()
    conn.close()
    return existe is not None


@app.route('/form', methods=['GET', 'POST'])
def index():
    if request.method == 'POST':
        email = request.form.get('email')
        if email_exists(email):
            flash("O email já está em uso. Não é possível cadastrar.")
            return redirect(url_for('index'))
        
        # 执行注册逻辑
        name = request.form.get('name')
        password = request.form.get('password')
        nascimento = request.form.get('dataNascimento')
        telefone = request.form.get('tel')
        rg = request.form.get('rg')
        cpf = request.form.get('cpf')
        endereco = request.form.get('adress')
        cidade = request.form.get('city')
        numero = request.form.get('num')
        uf = request.form.get('top')
        cep = request.form.get('cep')
        complemento = request.form.get('com')

        # 密码哈希处理
        hashed_password = generate_password_hash(password)
        
        conn = get_db_connection()
        if not conn:
            flash("数据库连接失败，请稍后重试")
            return redirect(url_for('index'))
        
        cursor = conn.cursor()
        sql = """INSERT INTO usuarios 
                 (name, email, password, nascimento, endereco, cidade, cpf, telefone, numero, uf, rg, cep, complemento) 
                 VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)"""
        valores = (name, email, hashed_password, nascimento, endereco, cidade, cpf, telefone, numero, uf, rg, cep, complemento)
        
        try:
            cursor.execute(sql, valores)
            conn.commit()
            flash("注册成功，请登录")
            return redirect(url_for('login_page'))
        except Error as e:
            print(f"注册错误: {e}")
            conn.rollback()
            flash("注册失败，请检查信息后重试")
        finally:
            cursor.close()
            conn.close()

    return render_template('registro-associado.html')


@app.route('/', methods=['GET', 'POST'])
def html():
    return render_template('index.html')


def login(name, email, password, cpf):
    conn = get_db_connection()
    if not conn:
        return None
    cursor = conn.cursor()
    query = "SELECT * FROM usuarios WHERE name = %s AND email = %s AND cpf = %s"
    cursor.execute(query, (name, email, cpf))
    user = cursor.fetchone()
    
    # 验证密码哈希
    if user and check_password_hash(user[3], password):
        cursor.close()
        conn.close()
        return user
    
    cursor.close()
    conn.close()
    return None


@app.route('/login', methods=['GET', 'POST'])
def login_page():
    if request.method == 'POST':
        name = request.form['name']
        email = request.form['email']
        password = request.form['password']
        cpf = request.form['cpf']

        user = login(name, email, password, cpf)

        if user:
            session['name'] = user[1]
            session['email'] = user[2]
            session['cpf'] = user[7]
            flash(f"Login bem-sucedido! Bem-vindo, {user[1]}", 'success')
            return redirect(url_for('cartao'))
        else:
            flash("Credenciais inválidas. Tente novamente.", 'error')

    return render_template('login.html')


def make_payment(client_secret):
    try:
        payment_intent = stripe.PaymentIntent.confirm(client_secret)
        if payment_intent.status == "succeeded":
            # 从session获取用户信息，避免重复登录验证
            if 'email' not in session or 'cpf' not in session:
                return False
            
            conn = get_db_connection()
            if not conn:
                return False
            
            cursor = conn.cursor()
            # 启用需要的更新逻辑
            # query = "UPDATE usuarios SET cartao = 1 WHERE email = %s AND cpf = %s"
            # cursor.execute(query, (session['email'], session['cpf']))
            conn.commit()
            cursor.close()
            conn.close()
            return True
        elif payment_intent.status == "requires_payment_method":
            # 处理支付失败逻辑
            return False
    except stripe.error.StripeError as e:
        print(f"Stripe支付错误: {e}")
        return False


# 移除启动时的自动调用，改为通过路由触发
@app.route('/payment', methods=['POST'])
def process_payment():
    client_secret = request.form.get('client_secret')
    if make_payment(client_secret):
        flash("支付成功")
    else:
        flash("支付失败，请重试")
    return redirect(url_for('cartao'))


@app.route('/logout')
def logout():
    session.clear()
    flash("Logout realizado com sucesso.", 'success')
    return redirect(url_for('html'))


@app.route('/index')
def page():
    if 'name' in session:
        return render_template('index.html', name=session['name'], email=session['email'], cpf=session['cpf'])
    else:
        return "Acesso negado. Faça o login primeiro."
    

@app.route('/cartao')
def cartao():
    if 'name' in session:
        return render_template('cartao.html', name=session['name'], email=session['email'], cpf=session['cpf'])
    else:
        return "Acesso negado. Faça o login primeiro."


if __name__ == '__main__':
    app.run(debug=True)

额外建议 #

• 使用数据库连接池提升性能，避免频繁创建/关闭连接。
• 为Stripe操作添加日志记录，便于排查支付问题。
• 完善表单验证逻辑，避免空值或非法数据进入数据库。

内容的提问来源于stack exchange，提问作者Davi