重构 Flask 服务端项目对于 SQL 的配置使用和延迟的请求回调巧妙设计运用

Posted 2022-10-28

一. Flask-SQLAlchemy

SQLAlchemy的声明扩展是使用SQLAlchemy的最新方法，可以像Django一样在一个位置定义表和模型，然后在任何地方使用。

from sqlalchemy import create_engine
from sqlalchemy.orm import scoped_session, sessionmaker
from sqlalchemy.ext.declarative import declarative_base

engine = create_engine(sqlite:////tmp/test.db)
db_session = scoped_session(sessionmaker(autocommit=False,
                                         autoflush=False,
                                         bind=engine))
Base = declarative_base()
Base.query = db_session.query_property()

def init_db():
    import yourapplication.models
    Base.metadata.create_all(bind=engine)

在这里导入定义模型所需要的所有模块，这样它们就会正确的注册在元数据上。否则你就必须在调用 init_db() 之前导入它们。

要定义模型，只需继承上面创建的Base类。你可能会好奇，这里为什么不关心线程。(与SQLite3一样，使用G对象。)原因是SQLAlchemy已经用scoped_session为我们做了这样的事情。 要在应用程序中声明性地使用SQLAlchemy，只需将以下代码添加到应用程序模块中：Flask在请求结束或应用程序终止时自动删除数据库会话：

from yourapplication.database import db_session

@app.teardown_appcontext
def shutdown_session(exception=None):
    db_session.remove()

人工对象关系映射

人工对象关系映射与上述声明方式相比有优点和缺点。主要区别在于手动对象关系映射分别定义和映射表和类。这种方法更灵活，但需要更多的代码。通常，这种方法的运行方式与声明相同，因此必须将应用程序划分到包中的多个模块中。

from sqlalchemy import create_engine, MetaData
from sqlalchemy.orm import scoped_session, sessionmaker

engine = create_engine(sqlite:////tmp/test.db)
metadata = MetaData()
db_session = scoped_session(sessionmaker(autocommit=False,
                                         autoflush=False,
                                         bind=engine))
def init_db():
    metadata.create_all(bind=engine)

二. 延迟的请求回调

Flask的设计思想之一是，在创建响应对象之后，它被传递给一系列回调函数，这些回调函数可以修改或替换响应对象。请求处理开始时，尚未创建响应对象。响应对象由视图函数或系统中的其他组件按需创建。 但是，当响应对象尚未创建时，我们如何修改它呢？例如，在request（）回调函数中，我们需要根据响应对象设置一个cookie。

from flask import request, after_this_request

@app.before_request
def detect_user_language():
    language = request.cookies.get(user_lang)

    if language is None:
        language = guess_language_from_request()

        # when the response exists, set a cookie with the language
        @after_this_request
        def remember_language(response):
            response.set_cookie(user_lang, language)
            return response

    g.language = language

通常我们选择避免这种情况。例如，可以尝试将应用程序逻辑移动到after_Therequest（）回调函数。然而，有时这种方法令人不快，或者使代码难看。 另一种方法是使用after_this_ request（）回调函数，该函数仅在当前请求之后执行。这允许您在应用程序中的任何位置延迟回调函数的执行。 在请求中的任何时候，您都可以注册将在请求结束时调用的函数。例如，下面的示例显示在before_ request（）回调函数会记住cookie中当前用户的语言：

基于 Celery 的后台任务

首先，需要有一个Celery实例，称为Celery应用程序。它的状态与Flask中的Flask相同。此实例用作所有与Celery相关的事务的条目，例如创建任务和管理工人，因此它必须由其他模块导入。 例如，可以将其放在任务模块中。通过这种方式，您可以使用任务的子类，添加对Flask应用程序场景的支持，并且无需重新配置即可挂接Flask配置。 酒窖可以在福尔斯克使用，只要：

from celery import Celery

def make_celery(app):
    celery = Celery(
        app.import_name,
        backend=app.config[CELERY_RESULT_BACKEND],
        broker=app.config[CELERY_BROKER_URL]
    )
    celery.conf.update(app.config)

    class ContextTask(celery.Task):
        def __call__(self, *args, **kwargs):
            with app.app_context():
                return self.run(*args, **kwargs)

    celery.Task = ContextTask
    return celery

此函数创建一个新的Celery对象，在应用程序配置中使用代理，并从Flask配置中更新剩余的Celey配置。然后创建任务子类，将任务执行打包到应用程序上下文中。

挂接，扩展

API文档可以在任何地方看到可用的重载、挂钩点和信号。您可以自定义自定义类，例如请求或响应对象。请深入研究您使用的API，以及哪些可定制部件可用于Flask分发。请研究哪些项目可以重构为工具集或Flask扩展。您可以在社区中找到许多扩展。如果你找不到满意的东西，就自己写一篇。

继承 Flask

from flask import Flask, Request
from werkzeug.datastructures import ImmutableOrderedMultiDict
class MyRequest(Request):
    """Request subclass to override request parameter storage"""
    parameter_storage_class = ImmutableOrderedMultiDict
class MyFlask(Flask):
    """Flask subclass using the custom request class"""
    request_class = MyRequest

以这种方式重载或者增强 Flask 的内部功能。