如何关闭在线程中使用的 Flask-SQLAlchemy 连接？

Question

当我在线程中查询 Flask-SQLAlchemy 模型时，数据库连接似乎从未关闭。这是一个测试用例：

from threading import Thread
from sqlalchemy.pool import NullPool
from flask_sqlalchemy import SQLAlchemy
from flask import Flask
db = SQLAlchemy()
app = Flask(__name__)

app.config['SQLALCHEMY_DATABASE_URI'] = 'postgresql:///testdb'
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
app.config['SQLALCHEMY_ENGINE_OPTIONS'] = {
    'poolclass': NullPool,
}
db.app = app
db.init_app(app)


class TestModel(db.Model):
    __tablename__ = "test_table"

    id = db.Column(db.Integer, autoincrement=True, primary_key=True)


@app.route('/test-connection')
def test_connection():
    def run_test():
        models = TestModel.query.all()
        print(models)
    thread = Thread(target=run_test)
    thread.start()
    return "OK"


app.run(debug=True, host='0.0.0.0')

当我运行我的路线时，它会idle in transaction在我的 pg_stat_activity 表中永久留下一个查询：

testdb=# select query, state from pg_stat_activity where query like '%test_table%' and query not like '%pg_stat_activity%';
                 query                  |        state        
----------------------------------------+---------------------
 SELECT test_table.id AS test_table_id +| idle in transaction
 FROM test_table                        | 
(1 row)

run_test如果我同步调用该函数而不是在线程中调用该函数，则该行不会显示。在生产中，这导致我的应用程序耗尽数据库连接并崩溃。在线程中使用 Flask-SQLAlchemy 数据库连接后如何关闭它？

我使用的是 Python 3.8.6、SQLAlchemy 1.3.18 和 Flask-SQLAlchemy 2.4.4。

Answer 1

我们在开放连接方面也遇到过类似的问题。尽管所讨论的数据库是 SQL Server，但我相信相同的解决方案应该适用于此。

我们添加了一个方法来关闭数据库连接。

def cleanup(session):
    """
    This method cleans up the session object and closes the connection pool using the dispose 
    method.
    """
    session.close()
    engine_container.dispose()

engine_container 定义为： engine_container = db.get_engine(app)

我们在每次请求后从finally 块中调用此方法。

finally:
    cleanup(db.session)

通过此更改，您的代码应如下所示：

from threading import Thread
from sqlalchemy.pool import NullPool
from flask_sqlalchemy import SQLAlchemy
from flask import Flask

db = SQLAlchemy()
app = Flask(__name__)

app.config['SQLALCHEMY_DATABASE_URI'] = 'postgresql:///testdb'
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
app.config['SQLALCHEMY_ENGINE_OPTIONS'] = {
    'poolclass': NullPool,
}
db.app = app
db.init_app(app)
engine_container = db.get_engine(app)


class TestModel(db.Model):
    __tablename__ = "test_table"

    id = db.Column(db.Integer, autoincrement=True, primary_key=True)


@app.route('/test-connection')
def test_connection():
    def run_test():
        try:
            models = TestModel.query.all()
            print(models)
        except Exception as err:
            raise err
        finally:
            cleanup(db.session)

    thread = Thread(target=run_test)
    thread.start()
    return "OK"


def cleanup(session):
    """
    This method cleans up the session object and also closes the connection pool using the dispose method.
    """
    session.close()
    engine_container.dispose()


app.run(debug=True, host='0.0.0.0')