Flask：Flask中使用SQLAlchemy会引发”SQLite objects created in a thread can only be used in that same thread”错误

在本文中，我们将介绍在使用Flask时，当我们尝试在多线程环境使用SQLAlchemy会引发的”SQLite objects created in a thread can only be used in that same thread”错误。我们将深入了解这个错误的原因，并提供解决方案和示例代码。

阅读更多：Flask 教程

了解错误原因

当在Flask应用程序中使用SQLAlchemy时，我们可能会遇到这样的场景：我们创建了一个线程，并在该线程中尝试使用SQLAlchemy的会话（session）对象，然后会抛出”SQLite objects created in a thread can only be used in that same thread”错误。这是因为SQLite引擎是单线程的，并且允许在同一线程中使用在该线程中创建的对象。

当我们创建一个新的线程时，我们实际上创建了一个新的SQLite连接和会话对象，并且在此线程中创建的这些SQLite对象不能在其他线程中重用。这就是为什么我们会遇到这个错误的原因。

解决方案

要解决这个问题，我们需要确保在每个线程中使用的SQLite连接和会话对象是独立的，而不是跨线程共享的。有几种方法可以实现这一点，在下面的示例代码中，我们将介绍两种解决方案。

解决方案一：使用Flask上下文管理器

一种常见的解决方案是使用Flask提供的上下文管理器，它可以自动处理线程间的对象共享问题。我们可以使用flask._app_ctx_stack来保存和获取当前的Flask应用上下文。

from flask import Flask, g
from flask_sqlalchemy import SQLAlchemy
import threading

app = Flask(__name__)

app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///database.db'
db = SQLAlchemy(app)


def get_db():
    if 'db' not in g:
        g.db = db.session

    return g.db


@app.teardown_appcontext
def teardown_db(exception=None):
    db = g.pop('db', None)

    if db is not None:
        db.close()


def task():
    with app.app_context():
        db = get_db()
        # 在这里使用db进行数据库操作

# 在线程中启动任务
t = threading.Thread(target=task)
t.start()

在上面的示例代码中，我们使用了flask._app_ctx_stack来保存和获取当前的Flask应用上下文。在任务函数中，我们首先使用get_db()函数来获取数据库会话对象。这个函数在每个线程中返回一个单独的会话对象，并将其保存在Flask应用上下文的”db”键中。在后面的数据库操作中，我们可以直接使用db对象。在任务完成后，我们可以使用teardown_db()函数来关闭数据库会话。

解决方案二：使用SQLAlchemy的scoped_session

另一种解决方案是使用SQLAlchemy提供的scoped_session。scoped_session会自动处理线程间的对象共享问题。

from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from sqlalchemy.orm import scoped_session, sessionmaker
import threading

app = Flask(__name__)

app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///database.db'
db = SQLAlchemy(app)
db_session = scoped_session(sessionmaker(bind=db.engine))


def task():
    session = db_session()
    # 在这里使用session进行数据库操作

# 在线程中启动任务
t = threading.Thread(target=task)
t.start()

在上面的示例代码中，我们首先创建了一个scoped_session对象db_session，它会自动管理线程间的数据库会话。在任务函数中，我们只需使用db_session来创建一个新的会话对象并在后续的数据库操作中使用。

示例代码

下面是一个完整的使用Flask和SQLAlchemy的示例代码，其中包含了上述解决方案：

from flask import Flask, g
from flask_sqlalchemy import SQLAlchemy
import threading

app = Flask(__name__)

app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///database.db'
db = SQLAlchemy(app)


def get_db():
    if 'db' not in g:
        g.db = db.session

    return g.db


@app.teardown_appcontext
def teardown_db(exception=None):
    db = g.pop('db', None)

    if db is not None:
        db.close()


def task():
    with app.app_context():
        db = get_db()
        # 在这里使用db进行数据库操作

# 解决方案二：使用SQLAlchemy的scoped_session
from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from sqlalchemy.orm import scoped_session, sessionmaker
import threading

app = Flask(__name__)

app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///database.db'
db = SQLAlchemy(app)
db_session = scoped_session(sessionmaker(bind=db.engine))


def task():
    session = db_session()
    # 在这里使用session进行数据库操作

# 在线程中启动任务
t = threading.Thread(target=task)
t.start()

总结

在本文中，我们介绍了在使用Flask时，当我们在多线程环境中使用SQLAlchemy会遇到的”SQLite objects created in a thread can only be used in that same thread”错误。我们了解了造成这个错误的原因，并提供了两种解决方案：使用Flask上下文管理器和使用SQLAlchemy的scoped_session。通过正确地处理数据库会话对象的线程间共享，我们可以避免这个错误。