扬州市邗江区城乡建设局网站,泗阳做网站,视频医疗平台网站开发,网站推广的基本方法对于大部分网站来说都是适用的有一些东西是大多数网络应用都会用到的。比如许多应用都会使用关系型数据库和用户 验证#xff0c;在请求之前连接数据库并得到当前登录用户的信息#xff0c;在请求之后关闭数据库连接。
更多用户贡献的代码片断和方案参见 current_app
主要内容#xff1a;
大型应用应用工…有一些东西是大多数网络应用都会用到的。比如许多应用都会使用关系型数据库和用户 验证在请求之前连接数据库并得到当前登录用户的信息在请求之后关闭数据库连接。
更多用户贡献的代码片断和方案参见 current_app
主要内容
大型应用应用工厂应用调度实现 API 异常处理URL 处理器使用 Distribute 部署使用 Fabric 部署在 Flask 中使用 SQLite 3在 Flask 中使用 SQLAlchemy上传文件缓存视图装饰器使用 WTForms 进行表单验证模板继承消息闪现通过 jQuery 使用 AJAX自定义出错页面惰性载入视图在 Flask 中使用 MongoKit添加一个页面图标流内容延迟的请求回调添加 HTTP 方法重载请求内容校验基于 Celery 的后台任务
大型应用
对于大型应用来说使用包代替模块是一个好主意。使用包非常简单。假设有一个小应用如下:
/yourapplication/yourapplication.py/static/style.css/templateslayout.htmlindex.htmllogin.html...
简单的包
要把上例中的小应用装换为大型应用只要在现有应用中创建一个名为 yourapplication 的新文件夹并把所有东西都移动到这个文件夹内。然后把 yourapplication.py 更名为 __init__.py 。请首先删除所有 .pyc 文件否则基本上会出问题
修改完后应该如下例:
/yourapplication/yourapplication/__init__.py/static/style.css/templateslayout.htmlindex.htmllogin.html...
但是现在如何运行应用呢原本的 python yourapplication/__init__.py 无法运行了。因为 Python 不希望包内的模块成为启动文件。但是这不是一个大问题只要在 yourapplication 文件夹旁添加一个 runserver.py 文件就可以了其内容如下:
from yourapplication import app
app.run(debugTrue)
我们从中学到了什么现在我们来重构一下应用以适应多模块。只要记住以下几点
Flask 应用对象必须位于 __init__.py文件中。这样每个模块就可以安全地导入了且 __name__变量会解析到正确的包。所有视图函数在顶端有 route() 的必须在 __init__.py 文件中被导入。不是导入对象本身而是导入视图模块。请 在应用对象创建之后导入视图对象。
__init__.py 示例:
from flask import Flask
app Flask(__name__)import yourapplication.views
views.py 内容如下:
from yourapplication import appapp.route(/)
def index():return Hello World!
最终全部内容如下:
/yourapplication/runserver.py/yourapplication/__init__.py/views.py/static/style.css/templateslayout.htmlindex.htmllogin.html...
回环导入
回环导入是指两个模块互相导入本例中我们添加的 views.py 就与 __init__.py 相互依赖。每个 Python 程序员都讨厌回环导入。一般情况下回环导入是个坏主意但 在这里一点问题都没有。原因是我们没有真正使用 __init__.py 中的视图只是 保证模块被导入并且我们在文件底部才这样做。
但是这种方式还是有些问题因为没有办法使用装饰器。要找到解决问题的灵感请参阅大型应用一节。
使用蓝图
对于大型应用推荐把应用分隔为小块每个小块使用蓝图辅助执行。关于这个主题的介绍 请参阅使用蓝图的模块化应用一节 。
应用工厂
如果你已经在应用中使用了包和蓝图 使用蓝图的模块化应用 那么还有许多方法可以更 进一步地改进你的应用。常用的方案是导入蓝图后创建应用对象但是如果在一个函数中创建对象那么就可以创建多个实例。
那么这样做有什么用呢
用于测试。可以针对不同的情况使用不同的配置来测试应用。用于多实例如果你需要运行同一个应用的不同版本的话。当然你可以在服务器上 使用不同配置运行多个相同应用但是如果使用应用工厂那么你可以只使用一个应用进程而得到多个应用实例这样更容易操控。
那么如何做呢
基础工厂
方法是在一个函数中设置应用具体如下:
def create_app(config_filename):app Flask(__name__)app.config.from_pyfile(config_filename)from yourapplication.model import dbdb.init_app(app)from yourapplication.views.admin import adminfrom yourapplication.views.frontend import frontendapp.register_blueprint(admin)app.register_blueprint(frontend)return app
这个方法的缺点是在导入时无法在蓝图中使用应用对象。但是你可以在一个请求中使用它。 如何通过配置来访问应用使用 a relnofollow hrefhttp://dormousehole.readthedocs.org/en/latest/api.html#flask.current_app stylepadding: 0px; margin: 0px; background-color: transparent; color: rgb(45, 133, 202);current_app:
from flask import current_app, Blueprint, render_template
admin Blueprint(admin, __name__, url_prefix/admin)admin.route(/)
def index():return render_template(current_app.config[INDEX_TEMPLATE])
这里我们在配置中查找模板的名称。
扩展对象初始化时不会绑定到一个应用应用可以使用 db.init_app 来设置扩展。 扩展对象中不会储存特定应用的状态因此一个扩展可以被多个应用使用。关于扩展设计的更多信息请参阅 Flask 扩展开发 。
当使用 Flask-SQLAlchemy 时你的 model.py 可能是这样的:
from flask.ext.sqlalchemy import SQLAlchemy
# no app object passed! Instead we use use db.init_app in the factory.
db SQLAlchemy()# create some models使用应用
因此要使用这样的应用就必须先创建它。下面是一个运行应用的示例 run.py 文件:
from yourapplication import create_app
app create_app(/path/to/config.cfg)
app.run()
改进工厂
上面的工厂函数还不是足够好可以改进的地方主要有以下几点
为了单元测试要想办法传入配置这样就不必在文件系统中创建配置文件。当设置应用时从蓝图调用一个函数这样就可以有机会修改属性如挂接请求前/后 处理器等。如果有必要的话当创建一个应用时增加一个 WSGI 中间件。
应用调度
应用调度是在 WSGI 层面组合多个 WSGI 应用的过程。可以组合多个 Flask 应用也可以 组合 Flask 应用和其他 WSGI 应用。通过这种组合如果有必要的话甚至可以在同一个 解释器中一边运行 Django 一边运行 Flask 。这种组合的好处取决于应用内部是如何 工作的。
应用调度与模块化的最大不同在于应用调度中的每个 应用是完全独立的它们以各自的配置运行并在 WSGI 层面被调度。
说明
下面所有的技术说明和举例都归结于一个可以运行于任何 WSGI 服务器的 application 对象。对于生产环境参见 部署方式 。对于开发环境 Werkzeug 提供了一个内建开发服务器它使用 werkzeug.serving.run_simple() 来运行:
from werkzeug.serving import run_simple
run_simple(localhost, 5000, application, use_reloaderTrue)
注意 werkzeug.wsgi.DispatcherMiddleware 。其原理是每个独立的 Flask 应用都 是一个合法的 WSGI 应用它们通过调度中间件组合为一个基于前缀调度的大应用。
假设你的主应用运行于 / 后台接口位于 /backend:
from werkzeug.wsgi import DispatcherMiddleware
from frontend_app import application as frontend
from backend_app import application as backendapplication DispatcherMiddleware(frontend, {/backend: backend
})
根据子域调度
有时候你可能需要使用不同的配置来运行同一个应用的多个实例。可以把应用创建过程 放在一个函数中这样调用这个函数就可以创建一个应用的实例具体实现参见应用工厂方案。
最常见的做法是每个子域创建一个应用配置服务器来调度所有子域的应用请求使用 子域来创建用户自定义的实例。一旦你的服务器可以监听所有子域那么就可以使用一个很简单的 WSGI 应用来动态创建应用了。
WSGI 层是完美的抽象层因此可以写一个你自己的 WSGI 应用来监视请求并把请求分配给你的 Flask 应用。如果被分配的应用还没有创建那么就会动态创建应用并被登记下来:
from threading import Lockclass SubdomainDispatcher(object):def __init__(self, domain, create_app):self.domain domainself.create_app create_appself.lock Lock()self.instances {}def get_application(self, host):host host.split(:)[0]assert host.endswith(self.domain), Configuration errorsubdomain host[:-len(self.domain)].rstrip(.)with self.lock:app self.instances.get(subdomain)if app is None:app self.create_app(subdomain)self.instances[subdomain] appreturn appdef __call__(self, environ, start_response):app self.get_application(environ[HTTP_HOST])return app(environ, start_response)
调度器示例:
from myapplication import create_app, get_user_for_subdomain
from werkzeug.exceptions import NotFounddef make_app(subdomain):user get_user_for_subdomain(subdomain)if user is None:# 如果子域没有对应的用户那么还是得返回一个 WSGI 应用# 用于处理请求。这里我们把 NotFound() 异常作为应用返回# 它会被渲染为一个缺省的 404 页面。然后可能还需要把# 用户重定向到主页。return NotFound()# 否则为特定用户创建应用return create_app(user)application SubdomainDispatcher(example.com, make_app)
根据路径调度
根据 URL 的路径调度非常简单。上面我们通过查找 Host 头来判断子域现在只要查找请求路径的第一个斜杠之前的路径就可以了:
from threading import Lock
from werkzeug.wsgi import pop_path_info, peek_path_infoclass PathDispatcher(object):def __init__(self, default_app, create_app):self.default_app default_appself.create_app create_appself.lock Lock()self.instances {}def get_application(self, prefix):with self.lock:app self.instances.get(prefix)if app is None:app self.create_app(prefix)if app is not None:self.instances[prefix] appreturn appdef __call__(self, environ, start_response):app self.get_application(peek_path_info(environ))if app is not None:pop_path_info(environ)else:app self.default_appreturn app(environ, start_response)
与根据子域调度相比最大的不同是根据路径调度时如果创建函数返回 None 那么就会回落到另一个应用:
from myapplication import create_app, default_app, get_user_for_prefixdef make_app(prefix):user get_user_for_prefix(prefix)if user is not None:return create_app(user)application PathDispatcher(default_app, make_app)
实现 API 异常处理
在 Flask 上经常会执行 RESTful API 。开发者首先会遇到的问题之一是用于 API 的内建异常处理不给力回馈的内容不是很有用。
对于非法使用 API 比使用 abort 更好的解决方式是实现你自己的异常处理类型 并安装相应句柄输出符合用户格式要求的出错信息。
简单的异常类
基本的思路是引入一个新的异常回馈一个合适的可读性高的信息、一个状态码和一些可选的负载给错误提供更多的环境内容。
以下是一个简单的示例:
from flask import jsonifyclass InvalidUsage(Exception):status_code 400def __init__(self, message, status_codeNone, payloadNone):Exception.__init__(self)self.message messageif status_code is not None:self.status_code status_codeself.payload payloaddef to_dict(self):rv dict(self.payload or ())rv[message] self.messagereturn rv
这样一个视图就可以抛出带有出错信息的异常了。另外还可以通过 payload 参数以字典的形式提供一些额外的负载。
注册一个错误处理句柄
现在视图可以抛出异常但是会立即引发一个内部服务错误。这是因为没有为这个错误处理类注册句柄。句柄增加很容易例如:
app.errorhandler(InvalidAPIUsage)
def handle_invalid_usage(error):response jsonify(error.to_dict())response.status_code error.status_codereturn response
在视图中的用法
以下是如何在视图中使用该功能:
app.route(/foo)
def get_foo():raise InvalidUsage(This view is gone, status_code410)
URL 处理器
New in version 0.7.
Flask 0.7 引入了 URL 处理器其作用是为你处理大量包含相同部分的 URL 。假设你有 许多 URL 都包含语言代码但是又不想在每个函数中都重复处理这个语言代码那么就可可以使用 URL 处理器。
在与蓝图配合使用时 URL 处理器格外有用。下面我们分别演示在应用中和蓝图中使用 URL 处理器。
国际化应用的 URL
假设有应用如下:
from flask import Flask, gapp Flask(__name__)app.route(/lang_code/)
def index(lang_code):g.lang_code lang_code...app.route(/lang_code/about)
def about(lang_code):g.lang_code lang_code...
上例中出现了大量的重复必须在每一个函数中把语言代码赋值给 g 对象。当然如果使用一个装饰器可以简化这个工作。但是当你需要生成由一个函数指向另一个函数的 URL 时还是得显式地提供语言代码相当麻烦。
我们使用 url_defaults() 函数来简化这个问题。这个函数可以自动 把值注入到 url_for() 。以下代码检查在 URL 字典中是否存在语言代码 端点是否需要一个名为 lang_code 的值:
app.url_defaults
def add_language_code(endpoint, values):if lang_code in values or not g.lang_code:returnif app.url_map.is_endpoint_expecting(endpoint, lang_code):values[lang_code] g.lang_code
URL 映射的 url_value_preprocessor() 。这些函数在请求匹配后立即根据 URL 的值执行代码。它们可以从 URL 字典中取出值并把取出的值放在 其他地方:
app.url_value_preprocessor
def pull_lang_code(endpoint, values):g.lang_code values.pop(lang_code, None)
这样就不必在每个函数中把 lang_code 赋值给 g 了。你还可以作 进一步改进写一个装饰器把语言代码作为 URL 的前缀。但是更好的解决方式是使用 蓝图。一旦 lang_code 从值的字典中弹出它就不再传送给视图函数了。精简后的代码如下:
from flask import Flask, gapp Flask(__name__)app.url_defaults
def add_language_code(endpoint, values):if lang_code in values or not g.lang_code:returnif app.url_map.is_endpoint_expecting(endpoint, lang_code):values[lang_code] g.lang_codeapp.url_value_preprocessor
def pull_lang_code(endpoint, values):g.lang_code values.pop(lang_code, None)app.route(/lang_code/)
def index():...app.route(/lang_code/about)
def about():...
国际化的蓝图 URL
因为蓝图可以自动给所有 URL 加上一个统一的前缀所以应用到每个函数就非常方便了。 更进一步因为蓝图 URL 预处理器不需要检查 URL 是否真的需要要一个 lang_code 参数所以可以去除 distribute 的前身是 setuptools 它是一个扩展库通常用于分发 Python 库和 扩展。它的英文名称的就是“分发”的意思。它扩展了 Python 自带的一个基础模块安装 系统 distutils 支持多种更复杂的结构方便了大型应用的分发部署。它的主要特色
支持依赖 一个库或者应用可以声明其所依赖的其他库的列表。依赖库将被自动 安装。包注册 可以在安装过程中注册包这样就可以通过一个包查询其他包的信息。 这套系统最有名的功能是“切入点”即一个包可以定义一个入口以便于其他包挂接 用以扩展包。安装管理 distribute 中的 easy_install 可以为你安装其他库。你也可以使用早晚会替代 easy_install 的 包 。 distribute 不支持分发标准模块因此我们不 讨论模块的问题。关于如何把模块转换为包的信息参见大型应用方案。使用 distribute 将使发布更复杂也更加自动化。如果你想要完全自动化处理请同时阅读使用 Fabric 部署一节。基础设置脚本因为你已经安装了 Flask 所以你应当已经安装了 setuptools 或 distribute 。如果 没有安装不用怕有一个 最好使用 virtualenv 。你的设置代码应用放在 setup.py 文件中这个文件应当位于应用旁边。这个文件名只是 一个约定但是最好不要改变因为大家都会去找这个文件。是的即使你使用 distribute 你导入的包也是 setuptools 。 distribute 完全向后兼容于 setuptools 因此它使用相同的导入名称。Flask 应用的基础 setup.py 文件示例如下:from setuptools import setupsetup(nameYour Application,version1.0,long_description__doc__,packages[yourapplication],include_package_dataTrue,zip_safeFalse,install_requires[Flask]
)请记住你必须显式的列出子包。如果你要 distribute 自动为你搜索包你可以使用 find_packages 函数:from setuptools import setup, find_packagessetup(...packagesfind_packages()
)大多数 setup 的参数可以望文生义但是 include_package_data 和 zip_safe 可以不容易理解。 include_package_data 告诉 distribute 要搜索一个 MANIFEST.in 文件把文件内容所匹配的所有条目作为包数据安装。可以通过使用这个参数分发 Python 模块的静态文件和模板参见分发资源 。 zip_safe 标志可用于强制或防止创建 zip 压缩包。通常你不会想要把包安装为 zip 压缩文件因为一些工具不支持压缩文件而且压缩文件比较难以调试。分发资源如果你尝试安装上文创建的包你会发现诸如 static 或 templates 之类的文件夹没有被安装。原因是 distribute 不知道要为你添加哪些文件。你要做的是在你的 setup.py 文件旁边创建一个 MANIFEST.in 文件。这个文件列出了所有应当添加到 tar 压缩包的文件:recursive-include yourapplication/templates *
recursive-include yourapplication/static *不要忘了把 setup 函数的 include_package_data 参数设置为 True 否则即使把内容在 MANIFEST.in 文件中全部列出来也没有用。声明依赖依赖是在 install_requires 参数中声明的这个参数是一个列表。列表中的每一项都是一个需要在安装时从 PyPI 获得的包。缺省情况下总是会获得最新版本的包但你可以指定最高版本和最低版本。示例:install_requires[Flask0.2,SQLAlchemy0.6,BrokenPackage0.7,1.0
]我前面提到依赖包都从 PyPI 获得的。但是如果要从别的地方获得包怎么办呢你只要 还是按照上述方法写然后提供一个可选地址列表就行了:dependency_links[http://example.com/yourfiles]请确保页面上有一个目录列表且页面上的链接指向正确的 tar 压缩包。这样 distribute 就会找到文件了。如果你的包在公司内部网络上请提供指向服务器的 URL 。安装 / 开发要安装你的应用理想情况下是安装到一个 virtualenv 只要运行带 install 参数的 setup.py 脚本就可以了。它会将你的应用安装到 virtualenv 的 site-packages 文件夹下同时下载并安装依赖:$ python setup.py install如果你正开发这个包同时也希望相关依赖被安装那么可以使用 develop 来代替:$ python setup.py develop这样做的好处是只安装一个指向 site-packages 的连接而不是把数据复制到那里。这样在开发过程中就不必每次修改以后再运行 install 了。使用 Fabric 部署Fabric 是一个 Python 工具与 Makefiles 类似但是能够在远程服务器上执行命令。如果与适当的 Python 包 大型应用 与优良的配置 配置管理 相结合那么 Fabric 将是在外部服务器上部署 Flask 的利器。在下文开始之前有几点需要明确Fabric 1.0 需要要被安装到本地。本教程假设使用的是最新版本的 Fabric 。应用已经是一个包且有一个可用的 setup.py 文件 使用 Distribute 部署 。在下面的例子中我们假设远程服务器使用 mod_wsgi 。当然你可以使用你自己 喜欢的服务器但是在示例中我们选择 Apache mod_wsgi 因为它们设置方便 且在没有 root 权限情况下可以方便的重载应用。创建第一个 Fabfilefabfile 是控制 Fabric 的东西其文件名为 fabfile.py 由 fab 命令执行。在这个文件中定义的所有函数都会被视作 fab 子命令。这些命令将会在一个或多个主机上运行。这些主机可以在 fabfile 中定义也可以在命令行中定义。本例将在 fabfile 中 定义主机。下面是第一个例子比较级。它可以把当前的源代码上传至服务器并安装到一个预先存在的 virtual 环境:from fabric.api import *# 使用远程命令的用户名
env.user appuser
# 执行命令的服务器
env.hosts [server1.example.com, server2.example.com]def pack():# 创建一个新的分发源格式为 tar 压缩包local(python setup.py sdist --formatsgztar, captureFalse)def deploy():# 定义分发版本的名称和版本号dist local(python setup.py --fullname, captureTrue).strip()# 把 tar 压缩包格式的源代码上传到服务器的临时文件夹put(dist/%s.tar.gz % dist, /tmp/yourapplication.tar.gz)# 创建一个用于解压缩的文件夹并进入该文件夹run(mkdir /tmp/yourapplication)with cd(/tmp/yourapplication):run(tar xzf /tmp/yourapplication.tar.gz)# 现在使用 virtual 环境的 Python 解释器来安装包run(/var/www/yourapplication/env/bin/python setup.py install)# 安装完成删除文件夹run(rm -rf /tmp/yourapplication /tmp/yourapplication.tar.gz)# 最后 touch .wsgi 文件让 mod_wsgi 触发应用重载run(touch /var/www/yourapplication.wsgi)上例中的注释详细应当是容易理解的。以下是 fabric 提供的最常用命令的简要说明run - 在远程服务器上执行一个命令local - 在本地机器上执行一个命令put - 上传文件到远程服务器上cd - 在服务器端改变目录。必须与 with 语句联合使用。运行 Fabfile那么如何运行 fabfile 呢答案是使用 fab 命令。要在远程服务器上部署当前版本的代码可以使用这个命令:$ fab pack deploy但是这个命令需要远程服务器上已经创建了 /var/www/yourapplication 文件夹且 /var/www/yourapplication/env 是一个 virtual 环境。更进一步服务器上还没有创建配置文件和 .wsgi 文件。那么我们如何在一个新的服务器上创建一个基础环境呢这个问题取决于你要设置多少台服务器。如果只有一台应用服务器多数情况下那么在 fabfile 中创建命令有一点多余。当然你可以这么做。这个命令可以称之为 setup 或 bootstrap 。在使用命令时显式传递服务器名称:$ fab -H newserver.example.com bootstrap设置一个新服务器大致有以下几个步骤在 /var/www 创建目录结构:$ mkdir /var/www/yourapplication
$ cd /var/www/yourapplication
$ virtualenv --distribute env上传一个新的 application.wsgi 文件和应用配置文件如 application.cfg 到服务器上。创建一个新的用于 yourapplication 的 Apache 配置并激活它。要确保激活 .wsgi 文件变动监视这样在 touch 的时候可以自动重载应用。 更多信息参见 flask.g 对象是绑定到请求的而不是应用环境。示例:app.route(/)
def index():cur get_db().cursor()...Note 请记住解散请求和应用环境的函数是一定会被执行的。即使请求前处理器执行失败或根本没有执行解散函数也会被执行。因此我们必须保证在关闭数据库连接之前 数据库连接是存在的。按需连接上述方式在第一次使用时连接数据库的优点是只有在真正需要时才打开数据库连接。 如果你想要在一个请求环境之外使用数据库连接那么你可以手动在 Python 解释器打开应用环境:with app.app_context():# now you can use get_db()简化查询现在在每个请求处理函数中可以通过访问 g.db 来得到当前打开的数据库连接。为了简化 SQLite 的使用这里有一个有用的行工厂函数。该函数会转换每次从数据库返回的 结果。例如为了得到字典类型而不是元组类型的返回结果可以这样:def make_dicts(cursor, row):return dict((cur.description[idx][0], value)for idx, value in enumerate(row))db.row_factory make_dicts或者更简单的:db.row_factory sqlite3.Row此外把得到游标执行查询和获得结果组合成一个查询函数不失为一个好办法:def query_db(query, args(), oneFalse):cur get_db().execute(query, args)rv cur.fetchall()cur.close()return (rv[0] if rv else None) if one else rv上述的方便的小函数与行工厂联合使用与使用原始的数据库游标和连接相比要方便多了。可以这样使用上述函数:for user in query_db(select * from users):print user[username], has the id, user[user_id]只需要得到单一结果的用法:user query_db(select * from users where username ?,[the_username], oneTrue)
if user is None:print No such user
else:print the_username, has the id, user[user_id]如果要给 SQL 语句传递参数请在语句中使用问号来代替参数并把参数放在一个列表中 一起传递。不要用字符串格式化的方式直接把参数加入 SQL 语句中这样会给应用带来 SQLAlchemy 来访问数据库。建议在你的 Flask 应用中使用包来代替 模块并把模型放入一个独立的模块中参见大型应用 。虽然这不是必须的但是很有用。有四种 SQLAlchemy 的常用方法下面一一道来Flask-SQLAlchemy 扩展因为 SQLAlchemy 是一个常用的数据库抽象层并且需要一定的配置才能使用因此我们为你做了一个处理 SQLAlchemy 的扩展。如果你需要快速的开始使用 SQLAlchemy 那么推荐你使用这个扩展。你可以从 Flask-SQLAlchemy 。声明SQLAlchemy 中的声明扩展是使用 SQLAlchemy 的最新方法它允许你像 Django 一样 在一个地方定义表和模型然后到处使用。除了以下内容我建议你阅读声明的官方文档。以下是示例 database.py 模块:from sqlalchemy import create_engine
from sqlalchemy.orm import scoped_session, sessionmaker
from sqlalchemy.ext.declarative import declarative_baseengine create_engine(sqlite:tmp/test.db, convert_unicodeTrue)
db_session scoped_session(sessionmaker(autocommitFalse,autoflushFalse,bindengine))
Base declarative_base()
Base.query db_session.query_property()def init_db():# 在这里导入定义模型所需要的所有模块这样它们就会正确的注册在# 元数据上。否则你就必须在调用 init_db() 之前导入它们。import yourapplication.modelsBase.metadata.create_all(bindengine)要定义模型的话只要继承上面创建的 Base 类就可以了。你可能会奇怪这里为什么不用理会线程就像我们在 SQLite3 的例子中一样使用 g 对象。 原因是 SQLAlchemy 已经用 scoped_session 为我们做好了此 类工作。如果要在应用中以声明方式使用 SQLAlchemy 那么只要把下列代码加入应用模块就可以了。 Flask 会自动在请求结束时或者应用关闭时删除数据库会话:from yourapplication.database import db_sessionapp.teardown_appcontext
def shutdown_session(exceptionNone):db_session.remove()以下是一个示例模型放入 models.py 中:from sqlalchemy import Column, Integer, String
from yourapplication.database import Baseclass User(Base):__tablename__ usersid Column(Integer, primary_keyTrue)name Column(String(50), uniqueTrue)email Column(String(120), uniqueTrue)def __init__(self, nameNone, emailNone):self.name nameself.email emaildef __repr__(self):return User %r % (self.name)可以使用 init_db 函数来创建数据库 from yourapplication.database import init_dbinit_db()在数据库中插入条目示例 from yourapplication.database import db_sessionfrom yourapplication.models import Useru User(admin, adminlocalhost)db_session.add(u)db_session.commit()查询很简单 User.query.all()
[User uadmin]User.query.filter(User.name admin).first()
User uadmin人工对象关系映射人工对象关系映射相较于上面的声明方式有优点也有缺点。主要区别是人工对象关系映射 分别定义表和类并映射它们。这种方式更灵活但是要多些代码。通常这种方式与声明 方式一样运行因此请确保把你的应用在包中分为多个模块。示例 database.py 模块:from sqlalchemy import create_engine, MetaData
from sqlalchemy.orm import scoped_session, sessionmakerengine create_engine(sqlite:tmp/test.db, convert_unicodeTrue)
metadata MetaData()
db_session scoped_session(sessionmaker(autocommitFalse,autoflushFalse,bindengine))
def init_db():metadata.create_all(bindengine)就像声明方法一样你需要在请求后或者应用环境解散后关闭会话。把以下代码放入你的应用模块:from yourapplication.database import db_sessionapp.teardown_appcontext
def shutdown_session(exceptionNone):db_session.remove()以下是一个示例表和模型放入 models.py 中:from sqlalchemy import Table, Column, Integer, String
from sqlalchemy.orm import mapper
from yourapplication.database import metadata, db_sessionclass User(object):query db_session.query_property()def __init__(self, nameNone, emailNone):self.name nameself.email emaildef __repr__(self):return User %r % (self.name)users Table(users, metadata,Column(id, Integer, primary_keyTrue),Column(name, String(50), uniqueTrue),Column(email, String(120), uniqueTrue)
)
mapper(User, users)查询和插入与声明方式的一样。SQL 抽象层如果你只需要使用数据库系统和 SQL 抽象层那么基本上只要使用引擎:from sqlalchemy import create_engine, MetaDataengine create_engine(sqlite:tmp/test.db, convert_unicodeTrue)
metadata MetaData(bindengine)然后你要么像前文中一样在代码中声明表要么自动载入它们:users Table(users, metadata, autoloadTrue)可以使用 insert 方法插入数据。为了使用事务我们必须先得到一个连接 con engine.connect()con.execute(users.insert(), nameadmin, emailadminlocalhost)SQLAlchemy 会自动提交。可以直接使用引擎或连接来查询数据库 users.select(users.c.id 1).execute().first()
(1, uadmin, uadminlocalhost)查询结果也是类字典元组 r users.select(users.c.id 1).execute().first()r[name]
uadmin你也可以把 SQL 语句作为字符串传递给 execute() 方法 engine.execute(select * from users where id :1, [1]).first()
(1, uadmin, uadminlocalhost)关于 SQLAlchemy 的更多信息请移步其save() 方法把文件永久 地保存在文件系统中。简介让我们从一个基本的应用开始这个应用上传文件到一个指定目录并把文件显示给用户。 以下是应用的引导代码:import os
from flask import Flask, request, redirect, url_for
from werkzeug import secure_filenameUPLOAD_FOLDER /path/to/the/uploads
ALLOWED_EXTENSIONS set([txt, pdf, png, jpg, jpeg, gif])app Flask(__name__)
app.config[UPLOAD_FOLDER] UPLOAD_FOLDER首先我们导入了一堆东西大多数是浅显易懂的。werkzeug.secure_filename() 会在稍后解释。UPLOAD_FOLDER 是上传文件要储存的目录ALLOWED_EXTENSIONS 是允许上传的文件扩展名的集合。接着我们给应用手动添加了一个 URL 规则。一般现在不会做这个但是为什么这么做了呢原因是我们需要服务器或我们的开发服务器为我们提供 服务。因此我们只生成这些文件的 URL 的规则。为什么要限制文件件的扩展名呢如果直接向客户端发送数据那么你可能不会想让用户上传任意文件。否则你必须确保用户不能上传 HTML 文件因为 HTML 可能引起 XSS 问题参见跨站脚本攻击XSS 。如果服务器可以执行 PHP 文件那么还必须确保不允许上传 .php 文件。但是谁又会在服务器上安装 PHP 呢对不 :)下一个函数检查扩展名是否合法上传文件把用户重定向到已上传文件的 URL:def allowed_file(filename):return . in filename and \filename.rsplit(., 1)[1] in ALLOWED_EXTENSIONSapp.route(/, methods[GET, POST])
def upload_file():if request.method POST:file request.files[file]if file and allowed_file(file.filename):filename secure_filename(file.filename)file.save(os.path.join(app.config[UPLOAD_FOLDER], filename))return redirect(url_for(uploaded_file,filenamefilename))return !doctype htmltitleUpload new File/titleh1Upload new File/h1form action methodpost enctypemultipart/form-datapinput typefile namefileinput typesubmit valueUpload/form那么 Plupload - HTML5, Java, FlashJumpLoader - Java
一个更简便的方案
因为所有应用中上传文件的方案基本相同因此可以使用 Flask-Uploads 扩展来实现 文件上传。这个扩展实现了完整的上传机制还具有白名单功能、黑名单功能以及其他 功能。
缓存
当你的应用变慢的时候可以考虑加入缓存。至少这是最简单的加速方法。缓存有什么用 假设有一个函数耗时较长但是这个函数在五分钟前返回的结果还是正确的。那么我们就 可以考虑把这个函数的结果在缓存中存放一段时间。
Flask 本身不提供缓存但是它的基础库之一 Werkzeug 有一些非常基本的缓存支持。它支持多种缓存后端通常你需要使用的是 memcached 服务器。
设置一个缓存
与创建 SimpleCache 对象。这个对象提供简单的缓存它把 缓存项目保存在 Python 解释器的内存中:
from werkzeug.contrib.cache import SimpleCache
cache SimpleCache()
如果你要使用 memcached 那么请确保有 memcache 模块支持你可以从 PyPI 获得模块和一个正在运行的 memcached 服务器。 连接 memcached 服务器示例:
from werkzeug.contrib.cache import MemcachedCache
cache MemcachedCache([127.0.0.1:11211])
如果你正在使用 App Engine 那么你可以方便地连接到 App Engine memcache 服务器:
from werkzeug.contrib.cache import GAEMemcachedCache
cache GAEMemcachedCache()
使用缓存
现在的问题是如何使用缓存呢有两个非常重要的操作 set() 。下面展示如何使用它们
get() 用于从缓存中获得项目调用时使用 一个字符串作为键名。如果项目存在那么就会返回这个项目否则返回 None:
rv cache.get(my-item)
set() 用于把项目添加到缓存中。第一个参数是键名第二个参数是键值。还可以提供一个超时参数当超过时间后项目会自动删除。
下面是一个完整的例子:
def get_my_item():rv cache.get(my-item)if rv is None:rv calculate_value()cache.set(my-item, rv, timeout5 * 60)return rv
视图装饰器
Python 有一个非常有趣的功能函数装饰器。这个功能可以使网络应用干净整洁。 Flask 中的每个视图都是一个装饰器它可以被注入额外的功能。你可以已经用过了 functools.wraps() 。
下面是检查登录装饰器的例子。假设登录页面为 login 当前用户被储存在 g.user 中如果还没有登录其值为 None:
from functools import wraps
from flask import g, request, redirect, url_fordef login_required(f):wraps(f)def decorated_function(*args, **kwargs):if g.user is None:return redirect(url_for(login, nextrequest.url))return f(*args, **kwargs)return decorated_function
如何使用这个装饰器呢把这个装饰器放在最靠近函数的地方就行了。当使用更进一步的装饰器时请记住要把 route() 装饰器放在最外面:
app.route(/secret_page)
login_required
def secret_page():pass
缓存装饰器
假设有一个视图函数需要消耗昂贵的计算成本因此你需要在一定时间内缓存这个视图的 计算结果。这种情况下装饰器是一个好的选择。我们假设你像缓存方案中一样设置了缓存。
下面是一个示例缓存函数。它根据一个特定的前缀实际上是一个格式字符串和请求的 当前路径生成缓存键。注意我们先使用了一个函数来创建装饰器这个装饰器用于装饰 函数。听起来拗口吧确实有一点复杂但是下面的示例代码还是很容易读懂的。
被装饰代码按如下步骤工作
基于基础路径获得当前请求的唯一缓存键。从缓存中获取键值。如果获取成功则返回获取到的值。否则调用原来的函数并把返回值存放在缓存中直至过期缺省值为五分钟。
代码:
from functools import wraps
from flask import requestdef cached(timeout5 * 60, keyview/%s):def decorator(f):wraps(f)def decorated_function(*args, **kwargs):cache_key key % request.pathrv cache.get(cache_key)if rv is not None:return rvrv f(*args, **kwargs)cache.set(cache_key, rv, timeouttimeout)return rvreturn decorated_functionreturn decorator
注意以上代码假设存在一个可用的实例化的 cache 对象更多信息参见缓存方案。
模板装饰器
不久前 TurboGear 的人发明了模板装饰器这个通用模式。其工作原理是返回一个字典 这个字典包含从视图传递给模板的值模板自动被渲染。以下三个例子的功能是相同的:
app.route(/)
def index():return render_template(index.html, value42)app.route(/)
templated(index.html)
def index():return dict(value42)app.route(/)
templated()
def index():return dict(value42)
正如你所见如果没有提供模板名称那么就会使用 URL 映射的端点把点转换为斜杠 加上 .html 。如果提供了那么就会使用所提供的模板名称。当装饰器函数返回 时返回的字典就被传送到模板渲染函数。如果返回的是 None 就会使用空字典。如果 返回的不是字典那么就会直接传递原封不动的返回值。这样就可以仍然使用重定向函数或 返回简单的字符串。
以下是装饰器的代码:
from functools import wraps
from flask import requestdef templated(templateNone):def decorator(f):wraps(f)def decorated_function(*args, **kwargs):template_name templateif template_name is None:template_name request.endpoint \.replace(., /) .htmlctx f(*args, **kwargs)if ctx is None:ctx {}elif not isinstance(ctx, dict):return ctxreturn render_template(template_name, **ctx)return decorated_functionreturn decorator
端点装饰器
当你想要使用 werkzeug 路由系统以便于获得更强的灵活性时需要和 Rule 中定义的一样把端点映射到视图函数。这样就需要 用的装饰器了。例如:
from flask import Flask
from werkzeug.routing import Ruleapp Flask(__name__)
app.url_map.add(Rule(/, endpointindex))app.endpoint(index)
def my_index():return Hello world
使用 WTForms 进行表单验证
当你必须处理浏览器提交的表单数据时视图代码很快会变得难以阅读。有一些库可以简化这个工作其中之一便是 WTForms 官方网站 。
模板继承
Jinja 最有力的部分就是模板继承。模板继承允许你创建一个基础“骨架”模板。这个模板中包含站点的常用元素定义可以被子模板继承的 块 。
听起来很复杂其实做起来简单看看下面的例子就容易理解了。
基础模板
这个模板的名称是 layout.html 它定义了一个简单的 HTML 骨架用于显示一个简单的两栏页面。“子”模板的任务是用内容填充空的块
!doctype html
htmlhead{% block head %}link relstylesheet href{{ url_for(static, filenamestyle.css) }}title{% block title %}{% endblock %} - My Webpage/title{% endblock %}/headbodydiv idcontent{% block content %}{% endblock %}/divdiv idfooter{% block footer %}copy; Copyright 2010 by a hrefhttp://domain.invalid/you/a.{% endblock %}/div/body
/html
在这个例子中 {% block %} 标记定义了四个可以被子模板填充的块。 block 标记告诉模板引擎这是一个可以被子模板重载的部分。
子模板
子模板示例
{% extends layout.html %}
{% block title %}Index{% endblock %}
{% block head %}{{ super() }}style typetext/css.important { color: #336699; }/style
{% endblock %}
{% block content %}h1Index/h1p classimportantWelcome on my awesome homepage.
{% endblock %}
这里 {% extends %} 标记是关键它告诉模板引擎这个模板“扩展”了另一个模板 当模板系统评估这个模板时会先找到父模板。这个扩展标记必须是模板中的第一个标记。 如果要使用父模板中的块内容请使用 {{ super() }} 。
消息闪现
一个好的应用和用户界面都需要良好的反馈。如果用户得不到足够的反馈那么应用最终会被用户唾弃。 Flask 的闪现系统提供了一个良好的反馈方式。闪现系统的基本工作方式 是在且只在下一个请求中访问上一个请求结束时记录的消息。一般我们结合布局模板来使用闪现系统。
简单的例子
以下是一个完整的示例:
from flask import Flask, flash, redirect, render_template, \request, url_forapp Flask(__name__)
app.secret_key some_secretapp.route(/)
def index():return render_template(index.html)app.route(/login, methods[GET, POST])
def login():error Noneif request.method POST:if request.form[username] ! admin or \request.form[password] ! secret:error Invalid credentialselse:flash(You were successfully logged in)return redirect(url_for(index))return render_template(login.html, errorerror)if __name__ __main__:app.run()
以下是实现闪现的 layout.html 模板
!doctype html
titleMy Application/title
{% with messages get_flashed_messages() %}{% if messages %}ul classflashes{% for message in messages %}li{{ message }}/li{% endfor %}/ul{% endif %}
{% endwith %}
{% block body %}{% endblock %}
以下是 index.html 模板
{% extends layout.html %}
{% block body %}h1Overview/h1pDo you want to a href{{ url_for(login) }}log in?/a
{% endblock %}
login 模板
{% extends layout.html %}
{% block body %}h1Login/h1{% if error %}p classerrorstrongError:/strong {{ error }}{% endif %}form action methodpostdldtUsername:ddinput typetext nameusername value{{request.form.username }}dtPassword:ddinput typepassword namepassword/dlpinput typesubmit valueLogin/form
{% endblock %}
闪现消息的类别
New in version 0.3.
闪现消息还可以指定类别如果没有指定那么缺省的类别为 message 。不同的 类别可以给用户提供更好的反馈。例如错误消息可以使用红色背景。
使用 flash() 函数可以指定消息的类别:
flash(uInvalid password provided, error)
模板中的 get_flashed_messages() 的 结果。这个功能有助于在不同位置显示不同类别的消息。
{% with errors get_flashed_messages(category_filter[error]) %}
{% if errors %}
div classalert-message block-message errora classclose href#×/aul{%- for msg in errors %}li{{ msg }}/li{% endfor -%}/ul
/div
{% endif %}
{% endwith %}
通过 jQuery 使用 AJAX
http://example.com/myapp 上呢在服务端这个问题不成为问题可以使用 url_for() 函数来得到答案。但是如果我们使用 jQuery 那么就不能硬码应用的路径只能使用动态路径。怎么办
一个简单的方法是在页面中添加一个 script 标记设置一个全局变量来表示应用的根 路径。示例
script typetext/javascript$SCRIPT_ROOT {{ request.script_root|tojson|safe }};
/script
在 Flask 0.10 版本以前版本中使用 |safe 是有必要的是为了使 Jinja 不要转义 JSON 编码的字符串。通常这样做不是必须的但是在 script 内部我们需要这么做。
进一步说明
在 HTML 中 script 标记是用于声明 CDATA 的也就是说声明的内容不会被解析。 之间的内容都会被作为脚本处理。这也意味着 在 script 标记之间不会存在任何 / 。在这里 |tojson 会正确处理问题 并为你转义斜杠 {{ |tojson|safe }} 会被渲染为 \/script 。
在 Flask 0.10 版本中更进了一步把所有 HTML 标记都用 unicode 转义了这样使 Flask 自动把 HTML 转换为安全标记。
JSON 视图函数
现在让我们来创建一个服务端函数这个函数接收两个 URL 参数两个需要相加的数字 然后向应用返回一个 JSON 对象。下面这个例子是非常不实用的因为一般会在客户端完成类似工作但这个例子可以简单明了地展示如何使用 jQuery 和 Flask:
from flask import Flask, jsonify, render_template, request
app Flask(__name__)app.route(/_add_numbers)
def add_numbers():a request.args.get(a, 0, typeint)b request.args.get(b, 0, typeint)return jsonify(resulta b)app.route(/)
def index():return render_template(index.html)
正如你所见我还添加了一个 index 方法来渲染模板。这个模板会按前文所述载入 jQuery 。模板中有一个用于两个数字相加的表单和一个触发服务端函数的链接。
注意这里我们使用了 示例源代码 。
自定义出错页面
Flask 有一个方便的 掌握应用错误。
出错处理器
一个出错处理器是一个函数就像一个视图函数一样。与视图函数不同之处在于出错处理器 在出现错误时被调用且传递错误。错误大多数是一个 errorhandler() 装饰器注册注册时应提供异常的 出代码。请记住 Flask 不会 为你设置出错代码因此请确保在返回响应时同时提供 HTTP 状态代码。
以下是一个处理 “404 Page Not Found” 异常的示例:
from flask import render_templateapp.errorhandler(404)
def page_not_found(e):return render_template(404.html), 404
示例模板
{% extends layout.html %}
{% block title %}Page Not Found{% endblock %}
{% block body %}h1Page Not Found/h1pWhat you were looking for is just not there.pa href{{ url_for(index) }}go somewhere nice/a
{% endblock %}
惰性载入视图
Flask 通常使用装饰器。装饰器简单易用只要把 URL 放在相应的函数的前面就可以了。 但是这种方式有一个缺点使用装饰器的代码必须预先导入否则 Flask 就无法真正找到你的函数。
当你必须快速导入应用时这就会成为一个问题。在 Google App Engine 或其他系统中 必须快速导入应用。因此如果你的应用存在这个问题那么必须使用集中 URL 映射。
add_url_rule() 加上应用前缀和点符号:
def url(url_rule, import_name, **options):view LazyView(yourapplication. import_name)app.add_url_rule(url_rule, view_funcview, **options)url(/, views.index)
url(/user/username, views.user)
有一件事情要牢记请求前和请求后处理器必须在第一个请求前导入。
其余的装饰器可以同样用上述方法改写。
在 Flask 中使用 MongoKit
现在使用文档型数据库来取代关系型数据库已越来越常见。本方案展示如何使用 MongoKit ,它是一个用于操作 MongoDB 的文档映射库。
本方案需要一个运行中的 MongoDB 服务器和已安装好的 MongoKit 库。
使用 MongoKit 有两种常用的方法下面逐一说明
声明
声明是 MongoKit 的缺省行为。这个思路来自于 Django 或 SQLAlchemy 的声明。
下面是一个示例 app.py 模块:
from flask import Flask
from mongokit import Connection, Document# configuration
MONGODB_HOST localhost
MONGODB_PORT 27017# create the little application object
app Flask(__name__)
app.config.from_object(__name__)# connect to the database
connection Connection(app.config[MONGODB_HOST],app.config[MONGODB_PORT])
如果要定义模型那么只要继承 MongoKit 的 Document 类就行了。如果你已经读过 SQLAlchemy 方案那么可以会奇怪这里为什么没有使用会话甚至没有定义一个 init_db 函数。一方面是因为 MongoKit 没有类似会话在东西。有时候这样会多写一点代码但会使它的速度更快。另一方面是因为 MongoDB 是无模式的。这就意味着可以在 插入数据的时候修改数据结构。 MongoKit 也是无模式的但会执行一些验证以确保数据的完整性。
以下是一个示例文档把示例内容也放入 app.py :
def max_length(length):def validate(value):if len(value) length:return Trueraise Exception(%s must be at most %s characters long % length)return validateclass User(Document):structure {name: unicode,email: unicode,}validators {name: max_length(50),email: max_length(120)}use_dot_notation Truedef __repr__(self):return User %r % (self.name)
在当前连接中注册用户文档
connection.register([User]) 上例展示如何定义模式命名结构和字符串最大长度验证器。上例中还使用了一个 MongoKit 中特殊的 use_dot_notation 功能。缺省情况下 MongoKit 的运作方式和 Python 的字典类似。但是如果 use_dot_notation 设置为 True 那么就可几乎像其他 ORM 一样使用点符号来分隔属性。
可以像下面这样把条目插入数据库中from yourapplication.database import connectionfrom yourapplication.models import Usercollection connection[test].usersuser collection.User()user[name] uadminuser[email] uadminlocalhostuser.save()
注意 MongoKit 对于列类型的使用是比较严格的。对于 name 和 email 列你都不能使用 str 类型应当使用 unicode 。
查询非常简单list(collection.User.find())
[User uadmin]collection.User.find_one({name: uadmin})
User uadmin
PyMongo 兼容层
如果你只需要使用 PyMongo 也可以使用 MongoKit 。在这种方式下可以获得最佳的性能。注意以下示例中没有 MongoKit 与 Flask 整合的内容整合的方式参见上文:
from MongoKit import Connectionconnection Connection()
使用 insert 方法可以插入数据。但首先必须先得到一个连接。这个连接类似于 SQL 界的表。collection connection[test].usersuser {name: uadmin, email: uadminlocalhost}collection.insert(user)
MongoKit 会自动提交。
直接使用集合查询数据库list(collection.find())
[{u_id: ObjectId(4c271729e13823182f000000), uname: uadmin, uemail: uadminlocalhost}]collection.find_one({name: uadmin})
{u_id: ObjectId(4c271729e13823182f000000), uname: uadmin, uemail: uadminlocalhost}
查询结果为类字典对象r collection.find_one({name: uadmin})r[email]
uadminlocalhost
关于 MongoKit 的更多信息请移步其send_from_directory() 函数来写一个视图:
import os
from flask import send_from_directoryapp.route(/favicon.ico)
def favicon():return send_from_directory(os.path.join(app.root_path, static),favicon.ico, mimetypeimage/vnd.microsoft.icon)
上例中的 MIME 类型可以省略浏览器会自动猜测类型。但是我们在例子中明确定义了 省去了额外的猜测反正这个类型是不变的。
上例会通过你的应用来提供图标如果可能的话最好配置你的专用服务器来提供图标 配置方法参见网页服务器的文档。
另见 Wikipedia 上的update_template_context() 以确保 更新被渲染的内容。这样模板遍历流内容。由于每次产生内容后服务器都会把内容 发送给客户端因此可能需要缓存来保存内容。我们使用了 rv.enable_buffering(size) 来进行缓存。 5 是一个比较明智的缺省值。
环境中的流内容
New in version 0.9.
注意当你生成流内容时请求环境已经在函数执行时消失了。 Flask 0.9 为你提供了 一点帮助让你可以在生成器运行期间保持请求环境:
from flask import stream_with_context, request, Responseapp.route(/stream)
def streamed_response():def generate():yield Hello yield request.args[name]yield !return Response(stream_with_context(generate()))
如果没有使用 RuntimeError 错误。
延迟的请求回调
Flask 的设计思路之一是响应对象创建后被传递给一串回调函数这些回调函数可以修改 或替换响应对象。当请求处理开始的时候响应对象还没有被创建。响应对象是由一个视图函数或者系统中的其他组件按需创建的。
但是当响应对象还没有创建时我们如何修改响应对象呢比如在一个请求前函数中我们需要根据响应对象设置一个 cookie 。
通常我们选择避开这种情形。例如可以尝试把应用逻辑移动到请求后函数中。但是有时候这个方法让人不爽或者让代码变得很丑陋。
变通的方法是把一堆回调函数贴到 g 对象上并且在请求结束时调用这些回调函数。这样你就可以在应用的任意地方延迟回调函数的执行。
装饰器
下面的装饰器是一个关键它在 g 对象上注册一个函数列表:
from flask import gdef after_this_request(f):if not hasattr(g, after_request_callbacks):g.after_request_callbacks []g.after_request_callbacks.append(f)return f
调用延迟的回调函数
至此通过使用 after_this_request 装饰器使得函数在请求结束时可以被调用。现在我们来实现这个调用过程。我们把这些函数注册为 after_request() 回调函数:
app.after_request
def call_after_request_callbacks(response):for callback in getattr(g, after_request_callbacks, ()):callback(response)return response
一个实例
现在我们可以方便地随时随地为特定请求注册一个函数让这个函数在请求结束时被调用。 例如你可以在请求前函数中把用户的当前语言记录到 cookie 中:
from flask import requestapp.before_request
def detect_user_language():language request.cookies.get(user_lang)if language is None:language guess_language_from_request()after_this_requestdef remember_language(response):response.set_cookie(user_lang, language)g.language languageLogo
Table Of Contents
添加 HTTP 方法重载
一些 HTTP 代理不支持所有 HTTP 方法或者不支持一些较新的 HTTP 方法例如 PACTH 。在这种情况下可以通过使用完全相反的协议用一种 HTTP 方法来“代理”另一种 HTTP 方法。
实现的思路是让客户端发送一个 HTTP POST 请求并设置 X-HTTP-Method-Override 头部为需要的 HTTP 方法例如 PATCH 。
通过 HTTP 中间件可以方便的实现:
class HTTPMethodOverrideMiddleware(object):allowed_methods frozenset([GET,HEAD,POST,DELETE,PUT,PATCH,OPTIONS])bodyless_methods frozenset([GET, HEAD, OPTIONS, DELETE])def __init__(self, app):self.app appdef __call__(self, environ, start_response):method environ.get(HTTP_X_HTTP_METHOD_OVERRIDE, ).upper()if method in self.allowed_methods:method method.encode(ascii, replace)environ[REQUEST_METHOD] methodif method in self.bodyless_methods:environ[CONTENT_LENGTH] 0return self.app(environ, start_response)
通过以下代码就可以与 Flask 一同工作了:
from flask import Flaskapp Flask(__name__)
app.wsgi_app HTTPMethodOverrideMiddleware(app.wsgi_app)
请求内容校验
请求数据会由不同的代码来处理或者预处理。例如 JSON 数据和表单数据都来源于已经 读取并处理的请求对象但是它们的处理代码是不同的。这样当需要校验进来的请求 数据时就会遇到麻烦。因此有时候就有必要使用一些 API 。
幸运的是可以通过包装输入流来方便地改变这种情况。
下面的例子演示在 WSGI 环境下读取和储存输入数据得到数据的 SHA1 校验:
import hashlibclass ChecksumCalcStream(object):def __init__(self, stream):self._stream streamself._hash hashlib.sha1()def read(self, bytes):rv self._stream.read(bytes)self._hash.update(rv)return rvdef readline(self, size_hint):rv self._stream.readline(size_hint)self._hash.update(rv)return rvdef generate_checksum(request):env request.environstream ChecksumCalcStream(env[wsgi.input])env[wsgi.input] streamreturn stream._hash
要使用上面的类你只要在请求开始消耗数据之前钩接要计算的流就可以了。按小心操作 request.form 或类似东西。例如 before_request_handlers 就应当小心不要操作。
用法示例:
app.route(/special-api, methods[POST])
def special_api():hash generate_checksum(request)# Accessing this parses the input streamfiles request.files# At this point the hash is fully constructed.checksum hash.hexdigest()return Hash was: %s % checksum
基于 Celery 的后台任务
Celery 是一个 Python 编写的是一个异步任务队列/基于分布式消息传递的作业队列。 以前它有一个 Flask 的集成但是从版本 3 开始它进行了一些内部的重构已经不需要这个集成了。本文主要说明如何在 Flask 中正确使用 Celery 。本文假设你 已经阅读过了其官方文档中的 Celery 入门
安装 Celery
Celery 在 Python 包索引 PyPI 上榜上有名因此可以使用 pip 或 easy_install 之类标准的 Python 工具来安装:
$ pip install celery
配置 Celery
你首先需要有一个 Celery 实例这个实例称为 celery 应用。其地位就相当于 Flask 中 Flask 一样。这个实例被用作所有 Celery 相关事务的入口例如创建 任务、管理工人等等。因此它必须可以被其他模块导入。
例如你可以把它放在一个 tasks 模块中。这样不需要重新配置你就可以使用 tasks 的子类增加 Flask 应用环境的支持并钩接 Flask 的配置。
只要如下这样就可以在 Falsk 中使用 Celery 了:
from celery import Celerydef make_celery(app):celery Celery(app.import_name, brokerapp.config[CELERY_BROKER_URL])celery.conf.update(app.config)TaskBase celery.Taskclass ContextTask(TaskBase):abstract Truedef __call__(self, *args, **kwargs):with app.app_context():return TaskBase.__call__(self, *args, **kwargs)celery.Task ContextTaskreturn celery
这个函数创建了一个新的 Celery 对象使用了应用配置中的 broker 并从 Flask 配置中升级了 Celery 的其余配置。然后创建了一个任务子类在一个应用环境中包装了任务执行。
最小的例子
基于上文以下是一个在 Flask 中使用 Celery 的最小例子:
from flask import Flaskapp Flask(__name__)
app.config.update(CELERY_BROKER_URLredis://localhost:6379,CELERY_RESULT_BACKENDredis://localhost:6379
)
celery make_celery(app)celery.task()
def add_together(a, b):return a b
这个任务现在可以在后台调用了result add_together.delay(23, 42)result.wait()
65
运行 Celery 工人
至此如果你已经按上文一步一步执行你会失望地发现你的 .wait() 不会真正返回。这是因为你还没有运行 celery 。你可以这样以工人方式运行 celery:
$ celery -A your_application worker
把 your_application 字符串替换为你创建 celery 对像的应用包或模块。
