如何让Flask SQLAlchemy重用db连接？
我似乎无法让我的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中使用应用程序工厂，我部分地从Flask-SQLAlchemy docs复制了这段代码。Elseware那些文档提到，在错误的上下文中使用数据库会导致连接泄漏-也许我做的初始化是错误的？
回答 3
Stack Overflow用户
发布于 2013-08-01 23:36:23
在阅读了SQLAlchemy文档并摆弄了一下db实例之后，我终于得到了解决方案。在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()Stack Overflow用户
发布于 2015-03-16 08:32:17
编辑: 2.4.3版本的Flask-SQLAlchemy中不推荐使用SQLALCHEMY_COMMIT_ON_TEARDOWN。您可以在这里看到更改说明，其中他们建议您自己调用.commit()：
我所做的就是注册我自己的app.after_request，如果响应状态码小于400，它就会调用.commit()。这要求您正确地构建应用程序，以确保响应状态代码< 400的HTTP事务应提交到数据库，但我认为这是一个很好的设计原则。
-以下是过时的答案
在最新版本的Flask-SQLAlchemy中，app.after_request会自动调用session.remove()。
另请参阅此处的SQLALCHEMY_COMMIT_ON_TEARDOWN设置：
https://pythonhosted.org/Flask-SQLAlchemy/config.html?highlight=sqlalchemy_commit_on_teardown
这也将自动提交事务。
Stack Overflow用户
发布于 2012-05-18 00:12:16
您知道，在每次test method之前和之后都会调用setUp and tearDown。从你的代码看，为了确保数据库为空，你似乎需要它们。
但是，也有setUpClass and tearDownClass，它为每个测试类调用一次。
我相信您可以拆分当前拥有的代码，并将与db-connection相关的部分移动到Class-level中，同时将与test-method相关的部分保留在需要的位置。
https://stackoverflow.com/questions/10613812
复制相似问题