flask - Psycopg2 和 Flask - 将连接绑定(bind)到 before_request 和 teardown_appcontext

Question

干杯，伙计们，
重构我的 Flask 应用程序我卡在将 db 连接绑定(bind)到 @app.before_request并在 @app.teardown_appcontext 关闭它.我正在使用普通的 Psycopg2 和应用程序工厂模式。

首先，我创建了一个在应用程序工厂中调用的函数，这样我就可以将@app 用作suggested by Miguel Grinberg here。 :

def create_app(test_config=None):
    app = Flask(__name__, instance_relative_config=True)

    --

    from shop.db import connect_and_close_db
    connect_and_close_db(app)

    --

    return app

然后我尝试了 http://flask.pocoo.org/docs/1.0/appcontext/#storing-data 上建议的这种模式:

def connect_and_close_db(app):

    @app.before_request
    def get_db_test():
        conn_string = "dbname=testdb user=testuser password=test host=localhost"
        if 'db' not in g:
            g.db = psycopg2.connect(conn_string)
        return g.db

    @app.teardown_appcontext
    def close_connection(exception):
        db = g.pop('db', None)

        if db is not None:
            db.close()

结果是:

TypeError: 'psycopg2.extensions.connection' object is not callable

任何人都知道发生了什么以及如何使其工作？

此外，我想知道一旦创建游标绑定(bind)到 before_request，我将如何访问连接对象以创建游标。 ?

Answer 1

这个解决方案可能远非完美，而且它并不是真正的 DRY。我欢迎评论或基于此的其他答案。

实现原始 psycopg2支持，你可能需要看看connection pooler .还有一个good guide关于如何使用 Flask 实现这一点。

基本思想是首先创建连接池。您希望在 flask 应用程序初始化时建立它(这可以在 python 解释器中或通过可能有几个的 gunicorn worker - 在这种情况下，每个 worker 都有自己的连接池)。我选择将返回的池存储在配置中:

from flask import Flask, g, jsonify

import psycopg2
from psycopg2 import pool

app = Flask(__name__)

app.config['postgreSQL_pool'] = psycopg2.pool.SimpleConnectionPool(1, 20,
    user = "postgres",
    password = "very_secret",
    host = "127.0.0.1",
    port = "5432",
    database = "postgres")

注意 SimpleConnectionPool 的前两个参数是 min & max连接。这是连接到您的数据库服务器的连接数，bwtween 1 & 20在这种情况下。

接下来定义一个 get_db功能:

def get_db():
    if 'db' not in g:
        g.db = app.config['postgreSQL_pool'].getconn()
    return g.db

SimpleConnectionPool.getconn()这里使用的方法只是从池中返回一个连接，我们将其分配给 g.db并返回。这意味着当我们调用 get_db()它在代码中的任何地方返回相同的连接，或者如果不存在则创建一个连接。不需要 before.context装饰师。

请定义您的拆卸功能:

@app.teardown_appcontext
def close_conn(e):
    db = g.pop('db', None)
    if db is not None:
        app.config['postgreSQL_pool'].putconn(db)

这在应用程序上下文被销毁时运行，并使用 SimpleConnectionPool.putconn()收起连接。

最后定义一个路由:

@app.route('/')
def index():
    db = get_db()
    cursor = db.cursor()

    cursor.execute("select 1;")
    result = cursor.fetchall()
    print (result)

    cursor.close()
    return jsonify(result)

此代码适用于我针对在 docker 容器中运行的 postgres 进行测试。可能应该改进的一些事情:

这个观点不是很干。也许您可以将其中的一些移到 get_db函数，因此它返回一个游标。 (!!!)

当python解释器退出时，你也应该找到关闭与app.config['postgreSQL_pool'].closeall的连接。

尽管测试了某种监视池的方法会很好，但是您可以在负载下观察池/数据库连接并确保池器按预期运行。

(!!!)在另一 block 土地上， sqlalchemy.scoped_session documentation解释了更多与此相关的事情，并提供了一些关于其“ session ”如何与请求相关的理论。他们以这样一种方式实现了它，您可以调用 Session.query('SELECT 1')如果 session 尚不存在，它将创建 session 。

编辑 : 这里是 gist使用您的应用程序工厂模式，并在评论中使用示例。