Flask CRUD中ADD方法报错UnboundLocalError：cursor未赋值即引用

问题根源 #

你的ADD方法触发UnboundLocalError: local variable 'cursor' referenced before assignment，核心原因是finally块中尝试关闭的cursor/conn变量可能并未被初始化：

• 当请求未携带JSON、JSON缺少指定字段，或建立数据库连接前出现异常时，cursor和conn根本没被赋值，此时执行cursor.close()就会触发未绑定变量的错误。
• 其他方法暂时正常，只是因为它们的conn/cursor初始化在try块最开头，未触发前置异常，但本质上也存在同样的隐患。

修复方案 #

方案一：安全手动管理连接与游标 #

提前初始化变量为None，在finally块中判断变量存在后再执行关闭操作，同时优化参数校验避免KeyError：

import pymysql
from app import app
from db_config import mysql
from flask import jsonify, flash, request
from werkzeug.security import generate_password_hash, check_password_hash
from waitress import serve

@app.route('/add', methods=['POST'])
def add_user():
    conn = None
    cursor = None
    try:
        _json = request.json
        # 先判断请求是否携带JSON
        if not _json:
            return not_found()
        # 使用get方法避免字段缺失时的KeyError
        _name = _json.get('name')
        _email = _json.get('email')
        _password = _json.get('pwd')
        
        if _name and _email and _password and request.method == 'POST':
            _hashed_password = generate_password_hash(_password)
            sql = "INSERT INTO tbl_user(user_name, user_email, user_password) VALUES(%s, %s, %s)"
            data = (_name, _email, _hashed_password,)
            
            conn = mysql.connect()
            cursor = conn.cursor()
            cursor.execute(sql, data)
            conn.commit()
            
            resp = jsonify('User added successfully!')
            resp.status_code = 200
            return resp
        else:
            return not_found()
    except Exception as e:
        print(e)
        return jsonify({'error': str(e)}), 500
    finally:
        # 先判断变量存在再执行关闭
        if cursor:
            cursor.close()
        if conn:
            conn.close()

# 其他接口同步修改，消除隐患
@app.route('/get')
def users():
    conn = None
    cursor = None
    try:
        conn = mysql.connect()
        cursor = conn.cursor(pymysql.cursors.DictCursor)
        cursor.execute("SELECT * FROM tbl_user")
        rows = cursor.fetchall()
        resp = jsonify(rows)
        resp.status_code = 200
        return resp
    except Exception as e:
        print(e)
        return jsonify({'error': str(e)}), 500
    finally:
        if cursor:
            cursor.close()
        if conn:
            conn.close()

@app.route('/get/<int:id>')
def user(id):
    conn = None
    cursor = None
    try:
        conn = mysql.connect()
        cursor = conn.cursor(pymysql.cursors.DictCursor)
        cursor.execute("SELECT * FROM tbl_user WHERE user_id=%s", id)
        row = cursor.fetchone()
        resp = jsonify(row)
        resp.status_code = 200
        return resp
    except Exception as e:
        print(e)
        return jsonify({'error': str(e)}), 500
    finally:
        if cursor:
            cursor.close()
        if conn:
            conn.close()

@app.route('/update', methods=['PUT'])
def update_user():
    conn = None
    cursor = None
    try:
        _json = request.json
        if not _json:
            return not_found()
        _id = _json.get('user_id')
        _name = _json.get('user_name')
        _email = _json.get('user_email')
        _password = _json.get('user_password')
        
        if _name and _email and _password and _id and request.method == 'PUT':
            _hashed_password = generate_password_hash(_password)
            sql = "UPDATE tbl_user SET user_name=%s, user_email=%s, user_password=%s WHERE user_id=%s"
            data = (_name, _email, _hashed_password, _id,)
            
            conn = mysql.connect()
            cursor = conn.cursor()
            cursor.execute(sql, data)
            conn.commit()
            
            resp = jsonify('User updated successfully!')
            resp.status_code = 200
            return resp
        else:
            return not_found()
    except Exception as e:
        print(e)
        return jsonify({'error': str(e)}), 500
    finally:
        if cursor:
            cursor.close()
        if conn:
            conn.close()

@app.route('/delete/<int:id>', methods=['DELETE'])
def delete_user(id):
    conn = None
    cursor = None
    try:
        conn = mysql.connect()
        cursor = conn.cursor()
        cursor.execute("DELETE FROM tbl_user WHERE user_id=%s", (id,))
        conn.commit()
        resp = jsonify('User deleted successfully!')
        resp.status_code = 200
        return resp
    except Exception as e:
        print(e)
        return jsonify({'error': str(e)}), 500
    finally:
        if cursor:
            cursor.close()
        if conn:
            conn.close()

# 补充404错误处理函数
@app.errorhandler(404)
def not_found(error=None):
    message = {
        'status': 404,
        'message': 'Not Found: ' + request.url,
    }
    resp = jsonify(message)
    resp.status_code = 404
    return resp

方案二：使用上下文管理器自动管理资源 #

通过with语句自动处理连接和游标的释放，无需手动调用close，代码更简洁安全：

import pymysql
from app import app
from db_config import mysql
from flask import jsonify, flash, request
from werkzeug.security import generate_password_hash, check_password_hash
from waitress import serve

@app.route('/add', methods=['POST'])
def add_user():
    try:
        _json = request.json
        if not _json:
            return not_found()
        _name = _json.get('name')
        _email = _json.get('email')
        _password = _json.get('pwd')
        
        if _name and _email and _password and request.method == 'POST':
            _hashed_password = generate_password_hash(_password)
            sql = "INSERT INTO tbl_user(user_name, user_email, user_password) VALUES(%s, %s, %s)"
            data = (_name, _email, _hashed_password,)
            
            # with语句自动管理连接，退出时自动关闭
            with mysql.connect() as conn:
                # with语句自动管理游标，退出时自动关闭
                with conn.cursor() as cursor:
                    cursor.execute(sql, data)
                    conn.commit()
            
            resp = jsonify('User added successfully!')
            resp.status_code = 200
            return resp
        else:
            return not_found()
    except Exception as e:
        print(e)
        return jsonify({'error': str(e)}), 500

# 其他接口同理修改，示例：
@app.route('/get')
def users():
    try:
        with mysql.connect() as conn:
            with conn.cursor(pymysql.cursors.DictCursor) as cursor:
                cursor.execute("SELECT * FROM tbl_user")
                rows = cursor.fetchall()
                resp = jsonify(rows)
                resp.status_code = 200
                return resp
    except Exception as e:
        print(e)
        return jsonify({'error': str(e)}), 500

# 其余接口（user、update_user、delete_user）按相同模式修改即可

关键优化点 #

1. 安全资源释放：方案一中通过初始化变量+存在性判断，避免未绑定变量错误；方案二中通过with自动管理资源，彻底省去手动关闭的操作。
2. 参数校验增强：使用dict.get()替代直接索引，避免字段缺失时的KeyError；先判断请求是否携带JSON，提前拦截无效请求。
3. 错误响应标准化：在异常块中返回JSON格式的错误信息，方便前端处理，而非仅打印到控制台。

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