without tutorial example postgresql sqlalchemy flask flask-sqlalchemy

postgresql - tutorial - sqlalchemy documentation



¿Cómo hago que Flask SQLAlchemy reutilice las conexiones db? (4)

Dado que las preguntas fueron hechas hace un año, creo que el OP debe haber resuelto sus problemas. Pero para quienquiera que haya estado aquí (como yo) intentando averiguar qué está pasando, aquí está mi mejor explicación:

Como dijo Van, el problema es, de hecho, con el caso de prueba que llama a setUp y tearDown para cada prueba. Aunque la conexión no se está filtrando exactamente de SQLAlchemy, sino que se debe a que cada prueba tiene su propio setUp , se setUp múltiples instancias de la aplicación: cada aplicación tendrá su propio grupo de conexión de base de datos, que presumiblemente no se reutiliza ni recicla cuando la prueba termina

En otras palabras, la conexión se está desprotegiendo y devolviendo al grupo de manera correcta , pero la conexión se mantiene como una conexión inactiva para futuras transacciones dentro de la misma aplicación (el punto de agrupación de conexiones).

En el caso de prueba anterior, se crean alrededor de 20 grupos de conexiones (cada uno con una conexión inactiva debido a create / drop_all) y ocupan el límite de conexión de Postgres.

Parece que no consigo que mi aplicación Flask cierre o reutilice las conexiones de base de datos. Estoy usando PostgreSQL 9.1.3 y

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

Cuando mi suite de pruebas ejecuta el número de conexiones abiertas, aumenta hasta que llega a 20 (la configuración de max_connections en postgresql.conf ), entonces veo:

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

He reducido el código al punto en el que solo está llamando a create_all y drop_all (pero no emitiendo ningún sql ya que no hay modelos).

Veo conexiones que se registran dentro y fuera en los registros:

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

Para cada ejecución de prueba, la dirección de la conexión (la parte "objeto de conexión en xyz") es diferente. Sospecho que esto tiene algo que ver con el problema, pero no estoy seguro de cómo investigar más.

El siguiente código reproduce el problema en un nuevo 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()

Esta es la primera vez que uso fábricas de aplicaciones en el matraz, y copié este código en parte de los documentos de Flask-SQLAlchemy . Elseware los documentos mencionan que usar una base de datos en el contexto incorrecto causará fugas en las conexiones, ¿tal vez estoy haciendo el inicio incorrectamente?

Después de leer los documentos de SQLAlchemy y algunos problemas con la instancia de db, finalmente obtuve la solución. Agregue db.get_engine(self.app).dispose() en tearDown() para que se vea como:

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

Usted sabe que el setUp and tearDown se llaman antes y después de cada test method . Desde su código parece que los necesita para garantizar una base de datos vacía.
Sin embargo, también hay setUpClass and tearDownClass , que se llaman una vez para cada clase de prueba.
Creo que puede dividir el código que tiene actualmente y mover la parte relacionada con db-connection al nivel de Class , mientras mantiene la parte relacionada con el test-method donde debe estar.