postgresql – 如何使Flask SQLAlchemy重用数据库连接?

前端之家收集整理的这篇文章主要介绍了postgresql – 如何使Flask SQLAlchemy重用数据库连接?前端之家小编觉得挺不错的,现在分享给大家,也给大家做个参考。
我似乎无法让我的Flask应用程序关闭或重新使用数据库连接。我使用Postgresql 9.1.3和
  1. Flask==0.8
  2. Flask-sqlAlchemy==0.16
  3. psycopg2==2.4.5

当我的测试套件运行打开的连接数爬到20(在postgresql.conf中的max_connections设置),然后我看到:

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

我已经将代码缩小到只需调用create_all和drop_all(但不会发出任何sql,因为没有模型)。

我看到在日志中检入和取出的连接:

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

对于每个测试运行,连接的地址(“xyz的连接对象”部分)是不同的。我怀疑这与问题有关,但我不知道如何进一步调查。

下面的代码在新的venv中再现了问题:

  1. from flask import Flask
  2. from flask.ext.sqlalchemy import sqlAlchemy
  3. from unittest import TestCase
  4.  
  5. import logging
  6. logging.basicConfig(level=logging.DEBUG)
  7. logging.getLogger('sqlalchemy.pool').setLevel(logging.DEBUG)
  8. logging.getLogger('sqlalchemy.engine').setLevel(logging.DEBUG)
  9. logging.getLogger('sqlalchemy.dialects').setLevel(logging.DEBUG)
  10. logging.getLogger('sqlalchemy.orm').setLevel(logging.DEBUG)
  11.  
  12.  
  13. db = sqlAlchemy()
  14.  
  15. def create_app(config=None):
  16. app = Flask(__name__)
  17. app.config.from_object(config)
  18. db.init_app(app)
  19. return app
  20.  
  21.  
  22. class AppTestCase(TestCase):
  23. sqlALCHEMY_DATABASE_URI = "postgresql://localhost/cx_test"
  24. TESTING = True
  25.  
  26. def create_app(self):
  27. return create_app(self)
  28.  
  29. def setUp(self):
  30. self.app = self.create_app()
  31. self.client = self.app.test_client()
  32. self._ctx = self.app.test_request_context()
  33. self._ctx.push()
  34. db.create_all()
  35.  
  36. def tearDown(self):
  37. db.session.remove()
  38. db.drop_all()
  39. self._ctx.pop()
  40.  
  41.  
  42. class TestModel(AppTestCase):
  43. def impl(self):
  44. logging.warn("impl")
  45. pass
  46.  
  47. def test_01(self):
  48. self.impl()
  49.  
  50. def test_02(self):
  51. self.impl()
  52.  
  53. def test_03(self):
  54. self.impl()
  55.  
  56. def test_04(self):
  57. self.impl()
  58.  
  59. def test_05(self):
  60. self.impl()
  61.  
  62. def test_06(self):
  63. self.impl()
  64.  
  65. def test_07(self):
  66. self.impl()
  67.  
  68. def test_08(self):
  69. self.impl()
  70.  
  71. def test_09(self):
  72. self.impl()
  73.  
  74. def test_10(self):
  75. self.impl()
  76.  
  77. def test_11(self):
  78. self.impl()
  79.  
  80. def test_12(self):
  81. self.impl()
  82.  
  83. def test_13(self):
  84. self.impl()
  85.  
  86. def test_14(self):
  87. self.impl()
  88.  
  89. def test_15(self):
  90. self.impl()
  91.  
  92. def test_16(self):
  93. self.impl()
  94.  
  95. def test_17(self):
  96. self.impl()
  97.  
  98. def test_18(self):
  99. self.impl()
  100.  
  101. def test_19(self):
  102. self.impl()
  103.  
  104.  
  105.  
  106. if __name__ == "__main__":
  107. import unittest
  108. unittest.main()

这是我第一次在烧瓶中使用应用程序工厂,我从Flask-SQLAlchemy docs.部分复制了这个代码Elseware这些文档提到在错误的上下文中使用数据库会导致连接泄漏 – 也许我正在做不正确的init?

在阅读sqlAlchemy文档和一些数据库实例之后,我终于得到了解决方案。在tearDown()中添加db.get_engine(self.app).dispose(),使其看起来像:
  1. def tearDown(self):
  2. db.session.remove()
  3. db.drop_all()
  4. db.get_engine(self.app).dispose()
  5. self._ctx.pop()

猜你在找的Postgre SQL相关文章