如何让 Flask SQLAlchemy 重用数据库连接？

Posted 2023-03-22

【中文标题】如何让 Flask SQLAlchemy 重用数据库连接？

【英文标题】：How do I make Flask SQLAlchemy reuse db connections?

【发布时间】：2012-05-23 17:40:49

【问题描述】：

我似乎无法让我的 Flask 应用程序关闭或重用数据库连接。我正在使用 PostgreSQL 9.1.3 和

Flask==0.8
Flask-SQLAlchemy==0.16
psycopg2==2.4.5

当我的测试套件运行时，打开的连接数不断攀升，直到达到 20（postgresql.conf 中的 max_connections 设置），然后我看到了：

OperationalError: (OperationalError) FATAL:  sorry, too many clients already
 None None

我已将代码缩减到只调用create_all 和drop_all 的程度（但没有发出任何sql，因为没有模型）。

我在日志中看到正在签入和签出的连接：

DEBUG:sqlalchemy.pool.QueuePool:Connection <connection object at 0x101c1dff0; dsn: 'dbname=cx_test host=localhost', closed: 0> checked out from pool
DEBUG:sqlalchemy.pool.QueuePool:Connection <connection object at 0x101c1dff0; dsn: 'dbname=cx_test host=localhost', closed: 0> being returned to pool
WARNING:root:impl   <-------- That's the test running
DEBUG:sqlalchemy.pool.QueuePool:Connection <connection object at 0x101c1dff0; dsn: 'dbname=cx_test host=localhost', closed: 0> checked out from pool
DEBUG:sqlalchemy.pool.QueuePool:Connection <connection object at 0x101c1dff0; dsn: 'dbname=cx_test host=localhost', closed: 0> being returned to pool

对于每个测试运行，连接的地址（“xyz 处的连接对象”部分）是不同的。我怀疑这与问题有关，但我不确定如何进一步调查。

下面的代码在一个新的 venv 中重现了这个问题：

from flask import Flask
from flask.ext.sqlalchemy import SQLAlchemy
from unittest import TestCase

import logging
logging.basicConfig(level=logging.DEBUG)
logging.getLogger('sqlalchemy.pool').setLevel(logging.DEBUG)
logging.getLogger('sqlalchemy.engine').setLevel(logging.DEBUG)
logging.getLogger('sqlalchemy.dialects').setLevel(logging.DEBUG)
logging.getLogger('sqlalchemy.orm').setLevel(logging.DEBUG)


db = SQLAlchemy()

def create_app(config=None):
    app = Flask(__name__)
    app.config.from_object(config)
    db.init_app(app)
    return app


class AppTestCase(TestCase):
    SQLALCHEMY_DATABASE_URI = "postgresql://localhost/cx_test"
    TESTING = True

    def create_app(self):
        return create_app(self)

    def setUp(self):
        self.app = self.create_app()
        self.client = self.app.test_client()
        self._ctx = self.app.test_request_context()
        self._ctx.push()
        db.create_all()

    def tearDown(self):
        db.session.remove()
        db.drop_all()
        self._ctx.pop()


class TestModel(AppTestCase):
    def impl(self):
        logging.warn("impl")
        pass

    def test_01(self):
        self.impl()

    def test_02(self):
        self.impl()

    def test_03(self):
        self.impl()

    def test_04(self):
        self.impl()

    def test_05(self):
        self.impl()

    def test_06(self):
        self.impl()

    def test_07(self):
        self.impl()

    def test_08(self):
        self.impl()

    def test_09(self):
        self.impl()

    def test_10(self):
        self.impl()

    def test_11(self):
        self.impl()

    def test_12(self):
        self.impl()

    def test_13(self):
        self.impl()

    def test_14(self):
        self.impl()

    def test_15(self):
        self.impl()

    def test_16(self):
        self.impl()

    def test_17(self):
        self.impl()

    def test_18(self):
        self.impl()

    def test_19(self):
        self.impl()



if __name__ == "__main__":
    import unittest
    unittest.main()

这是我第一次在烧瓶中使用应用工厂，我从Flask-SQLAlchemy docs 复制了部分代码。 Elseware 那些文档提到在错误的上下文中使用数据库会导致连接泄漏 - 也许我的初始化不正确？

【参考方案1】：

在阅读了 SQLAlchemy 文档并摆弄了一些数据库实例之后，我终于得到了解决方案。在tearDown() 中添加db.get_engine(self.app).dispose() 使其看起来像：

def tearDown(self):
    db.session.remove()
    db.drop_all()
    db.get_engine(self.app).dispose()
    self._ctx.pop()

【参考方案2】：

自从大约一年前提出问题以来，我认为 OP 一定已经解决了他的问题。但对于那些徘徊到这里（像我一样）试图弄清楚发生了什么的人来说，这是我最好的解释：

正如 van 所说，问题确实出在每个测试调用 setUp 和 tearDown 的测试用例上。虽然连接并没有完全从 SQLAlchemy 泄漏，而是因为每个测试都有自己的setUp，因此创建了应用程序的多个实例：每个应用程序都有自己的数据库连接池，可能不会被重用或在测试完成时回收。

换句话说，连接被检出并正确返回到池中，但是连接作为空闲连接继续存在，用于同一内的未来事务应用程序（连接池的点）。

在上面的测试用例中，创建了大约 20 个连接池（每个都有空闲连接，因为 create/drop_all）并占用了 postgres 连接限制。

【参考方案3】：

编辑：SQLALCHEMY_COMMIT_ON_TEARDOWN 在 Flask-SQLAlchemy 版本 2.4.3 中已弃用。您可以在此处查看更改说明，他们建议您自行致电 .commit()：

https://flask-sqlalchemy.palletsprojects.com/en/2.x/changelog/?highlight=sqlalchemy_commit_on_teardown#version-2-4-3

我所做的是注册我自己的app.after_request，如果响应状态代码 .commit()。这需要您正确构建应用程序以确保响应状态代码

----下面的旧答案----

在最新版本的 Flask-SQLAlchemy 中，session.remove() 会自动在 app.after_request 中调用。

另外，请参阅此处的 SQLALCHEMY_COMMIT_ON_TEARDOWN 设置：

https://pythonhosted.org/Flask-SQLAlchemy/config.html?highlight=sqlalchemy_commit_on_teardown

这也会自动提交事务。

【参考方案4】：

您知道setUp and tearDown 在每个test method 之前和之后都会被调用。从您的代码看来，您需要它们以确保数据库为空。 不过也有setUpClass and tearDownClass，每个测试类调用一次。 我相信您可以拆分您当前拥有的代码并将db-connection 相关部分移动到Class 级别，同时将test-method 相关部分保留在需要的位置。

【讨论】：

谢谢 Van - 这解决了测试套件的问题，但这不只是意味着我泄漏的连接更少了吗？

@TomDunham：我想你是对的。没有postgres，所以无法为您提供帮助，抱歉..