一. 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 的内部功能。