尽管应用程序已初始化，但 Flask 在应用程序上下文之外工作

Question

我正在尝试使用 Flask 中的 MySQLdb 查询简单的 MySQL DB。我有以下代码：

from flask import Flask, render_template
import json
from flask_mysqldb import MySQL

app = Flask(__name__)

mysql = MySQL()

app.config['MYSQL_DATABASE_USER'] = 'dbuser'
app.config['MYSQL_DATABASE_PASSWORD'] = 'xxxxxxxxxxxxx'
app.config['MYSQL_DATABASE_DB'] = 'calendar'
app.config['MYSQL_DATABASE_HOST'] = 'localhost'
mysql.init_app(app)

db = mysql.connect()
cur = db.cursor()

当我运行此代码时，我在 Apache 日志中收到以下错误：

RuntimeError: working outside of application context

我已经验证在对 MySQLdb 进行任何调用之前我正在创建我的

app

对象并阅读许多页面/文档，但我无法理解为什么我的代码失败，因为它看起来与其他示例相同（例如 https://codehandbook.org/python-web-application-flask-mysql/ 和 https://code.tutsplus.com/tutorials/creating-a-web-app-from-scratch-using-python-flask -和-mysql--cms-22972）

Answer 1

mysql.connect() 到处都使用

current_app

，所以你必须在应用程序上下文中调用它。

方法一：临时推送应用上下文

with app.app_context():
    cur = mysql.connection.cursor()

通常情况下，你不需要创建全局对象。相反，您可以在视图函数内创建

cur

对象。

方法二：在视图函数内部调用

@app.route('/')
def users():
    cur = mysql.connection.cursor()
    cur.execute('''SELECT user, host FROM mysql.user''')
    rv = cur.fetchall()
    return str(rv)

如果你有一堆视图函数使用这个对象，你可以创建一个请求处理程序来创建它。

方法三：在

before_request

函数中调用，即可在所有视图函数中直接访问

cur

@app.before_request
def make_db():
    cur = mysql.connection.cursor()

Answer 2

而不是

mysql = MySQL()

使用

mysql = MySQL(app)

这样你的 mysql 对象就会连接到你的应用程序。

（您可以从这里开始了解这意味着什么：https://github.com/admiralobvious/flask-mysqldb/blob/master/flask_mysqldb/init.py#L8）

Answer 3

如果您正在使用

from flask_mysqldb import MySQL

使用。例如，我们将此页面命名为app_file.py

    app.config['MYSQL_HOST'] = "host_ip"
    app.config['MYSQL_USER'] = "your_username"
    app.config['MYSQL_PASSWORD'] = "your_pass"
    app.config['MYSQL_DB'] = "your_db"
    app.config['MYSQL_CURSORCLASS'] = "DictCursor"
    with app.app_context():
        db = MySQL(app)

现在您可以访问该变量了。下面是从不同类调用的示例。这里我使用的是蓝图。

from app_file.py import db
@yourapi.route("/testUrl")
def testMysql():
    cur = db.connection.cursor()
    cur.execute('''select * from user limit 0,10''')
    row_headers=[x[0] for x in cur.description] #this will extract row headers
    rv = cur.fetchall()
    json_data=[]
    for result in rv:
        json_data.append(dict(zip(row_headers,result)))
    return json.dumps(json_data)