连接到 Flask 中的数据库，哪种方法更好？ [关闭]

Posted 2023-02-23

【中文标题】连接到 Flask 中的数据库，哪种方法更好？ [关闭]

【英文标题】：Connect to a Database in Flask, Which Approach is better? [closed]

【发布时间】：2013-04-25 02:04:35

【问题描述】：

方法一：使用来自http://flask.pocoo.org/docs/tutorial/dbcon/和http://flask.pocoo.org/docs/patterns/sqlite3/的特殊g对象

import sqlite3
from flask import g

DATABASE = '/path/to/database.db'

def connect_db():
    return sqlite3.connect(DATABASE)

@app.before_request
def before_request():
    g.db = connect_db()

@app.teardown_request
def teardown_request(exception):
    if hasattr(g, 'db'):
        g.db.close()

方法二：使用来自https://github.com/mitsuhiko/flask/blob/master/examples/flaskr/flaskr.py的神秘_app_ctx_stack

from sqlite3 import dbapi2 as sqlite3
from flask import Flask, request, session, g, redirect, url_for, abort, \
     render_template, flash, _app_ctx_stack
def get_db():
    """Opens a new database connection if there is none yet for the
    current application context.
    """
    top = _app_ctx_stack.top
    if not hasattr(top, 'sqlite_db'):
        top.sqlite_db = sqlite3.connect(app.config['DATABASE'])
    return top.sqlite_db


@app.teardown_appcontext
def close_db_connection(exception):
    """Closes the database again at the end of the request."""
    top = _app_ctx_stack.top
    if hasattr(top, 'sqlite_db'):
        top.sqlite_db.close()

哪种方法更好？有什么区别？

【问题讨论】：

相关，但也没有解释：flask.pocoo.org/docs/extensiondev

【参考方案1】：

两者的区别在于，无论您是否需要，方法一都会在 g.db 上创建连接，而方法二仅在您在该应用程序上下文中第一次调用 get_db 时创建连接。

如果你比较两者，使用这个设置：

yourapp = Flask(__name__)

# setup g.db or app_context here
# Add a logging statement (print will do)
# to the get_db or before_request functions
# that simply says "Getting the db connection ..."
# Then access / and /1

@yourapp.route("/")
def index():
    return "No database calls here!"

@yourapp.route("/<int:post_id>")
def show_post(post_id):
    # get a post using g.db or get_db
    return "Went to the DB and got !r".format(post)

您会看到，当您使用 @app.before_request 设置 (g.db) 点击 / 时，您会获得一个连接无论您是否使用它，同时使用 _app_context 路由只有当您致电get_db时，您才会获得连接。

公平地说，您还可以向g 添加一个描述符，该描述符将执行相同的延迟连接（或在现实生活中，从连接池获取连接）。在两种情况下，您可以使用更多的魔法（准确地说是werkzeug.local.LocalProxy）来创建您自己的自定义thread local，其行为类似于g、current_app 和request (@ 987654322@)。

【参考方案2】：

第一个问题是即使在不需要时也要获取连接。第二个缺点是使用第三方框架的内部结构，而且非常不可读。

仅在两者中，第二个可能是更好的选择。它不仅不会为不需要的路由获取连接，而且如果您沿着不需要的任何代码路径走下去，即使路由中的其他代码路径需要连接，它也不会获取连接。 （例如，如果您有一些表单验证，则仅在验证通过时才需要连接；验证失败时不会打开连接。）您只能在使用此设置之前获取连接。

但是，您可以避免弄乱内部结构，并且仍然可以获得所有这些好处。就个人而言，我创建了自己的小全局方法：

import flask
import sqlite3

def request_has_connection():
    return hasattr(flask.g, 'dbconn')

def get_request_connection():
    if not request_has_connection():
        flask.g.dbconn = sqlite3.connect(DATABASE)
        # Do something to make this connection transactional.
        # I'm not familiar enough with SQLite to know what that is.
    return flask.g.dbconn

@app.teardown_request
def close_db_connection(ex):
    if request_has_connection():
        conn = get_request_connection()
        # Rollback
        # Alternatively, you could automatically commit if ex is None
        # and rollback otherwise, but I question the wisdom 
        # of automatically committing.
        conn.close()

然后，在整个应用程序中，始终通过get_request_connection 获取您的连接，就像您使用get_db 函数一样。简洁高效。基本上，两全其美。

编辑：

回想起来，我真的不喜欢这些是全局方法这一事实，但我认为这是因为 Flask 就是这样工作的：它为您提供了实际上指向线程局部变量的“全局变量”。

【参考方案3】：

我推荐Flask-SQLAlchemy，它扩展了 SQLAlchemy 以在 Flask 中使用，因此它支持许多不同的数据库。 （来自 Flask-SQLAlchemy 文档的示例）

设置：

from flask import Flask
from flask.ext.sqlalchemy import SQLAlchemy

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:////tmp/test.db'
db = SQLAlchemy(app)


class User(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(80), unique=True)
    email = db.Column(db.String(120), unique=True)

    def __init__(self, username, email):
        self.username = username
        self.email = email

    def __repr__(self):
        return '<User %r>' % self.username

现在，您只需导入/使用 User 类即可访问数据库中的 User 表。

创建新用户：

>>> from yourapplication import User
>>> admin = User('admin', '[e-mail]')
>>> guest = User('guest', '[e-mail]')

将用户添加到数据库：

>>> db.session.add(admin)
>>> db.session.add(guest)
>>> db.session.commit()

查询数据库中已有的用户：

>>> users = User.query.all()
[<User u'admin'>, <User u'guest'>]
>>> admin = User.query.filter_by(username='admin').first()
<User u'admin'>

【讨论】：

你是对的。我想我还假设答案应该显示何时断开/关闭连接。您将如何实施连接/断开连接？ Method One: Using special g object 或 Method Two: Using Mysterious _app_ctx_stack.

@jberger 好点；我的回答并没有直接回答关于这两种方法中哪一种更好的原始问题，而是提供了完全不同的第三种选择。

【参考方案4】：

我会使用方法一 - 更具可读性和更少“hackish”。

方法 2 可能是为烧瓶扩展集成而设计的（example 和 explanation of app-ctx-stack）。虽然它们可能具有非常相似的效果，但方法一应该用于正常情况。

【参考方案5】：

要在同一个烧瓶会话中保留数据库连接，您可以使用应用程序上下文并分配数据库连接。 如果连接中断，上下文将在不断轮询连接对象时尝试重新建立连接。

from flask import Flask
application = Flask(__name__)


def connect_to_database():
  db_handler = SqliteDBConnect("uid=0;""pwd=1".format(UID, PWD),
                                table_prefix="".format(TBL_PRFX))
  return db_handler

fd = '_database': None
def get_db():
  db = fd['_database']
  if not isinstance(db, SqliteDBConnect):
    fd['_database'] = connect_to_database()
    db = fd['_database']
return db

with application.app_context():
  #Get DB connection from application's context
  db = LocalProxy(lambda: get_db())