How do I make Flask SQLAlchemy reuse db connections?

Question

I can't seem to get my Flask app to close or reuse DB connections. I'm using PostgreSQL 9.1.3 and

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

As my test suite runs the number of open connections climbs until it hits 20 (the max_connections setting in postgresql.conf), then I see:

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

I've reduced the code to the point where it's just calling create_all and drop_all (but not issuing any sql as there are no models).

I see connections being checked in and out in the logs:

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 

For each test run the address of the connection (the "connection object at xyz" part) is different. I suspect this has something to do with the problem, but I'm not sure how to investigate further.

The code below reproduces the problem in a new 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() 

This is the first time I've used app factories in flask, and I copied this code partly from the Flask-SQLAlchemy docs. Elseware those docs mention that using a db in the wrong context will cause connections to leak - maybe I am doing the init incorrectly?

Answer 1

After reading the SQLAlchemy docs and some fiddling with the db instance, I finally got the solution. Add db.get_engine(self.app).dispose() in tearDown() so that it looks like:

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