本篇所有操作均在基于 Ubuntu 16.04 LTS 的虚拟机下完成,且使用 Vagrant 来操作虚拟机系统,虚拟机系统 VirtualBox Version: 7.0
一、上下文
环境准备:
Python 2.7.11+
pip==9.0.3
flask==0.11.1
httpie==0.9.4
werkzeug==0.11.10
应用上下文的典型应用场景是缓存一些在发生请求之前要使用到的资源,比如生成数据库连接和缓存一些对象。
请求上下文发生在 HTTP 请求开始,WSGI Server 调用 Flask.__call__()
之后。
应用上下文并不是应用启动后生成的唯一上下文,应用上下文和请求上下文的关系如下:
1
2
3
4
5
6
7
8
9
10
11
class RequestContext(object):
self._implicit_app_ctx_stack = []
def push(self):
# some stuff
app_ctx = _app_ctx_stack.top
if app_ctx is None or app_ctx.app != self.app:
app_ctx = self.app.app_context()
app_ctx.push()
self._implicit_app_ctx_stack.append(app_ctx)
# some other stuff
也就是说应用上下文是被动的在推入请求上下文的过程中生成的,在请求结束的时候,也会把请求上下文弹出:
1
2
3
4
5
6
7
class RequestContext(object):
def pop(self, exc=_sentinel):
app_ctx = self._implicit_app_ctx_stack.pop()
try:
# some stuff
if app_ctx is not None:
app_ctx.pop(exc)
也就是说,事实上在 Web 应用环境中,请求上下文和应用上下文是一一对应的。请求上下文和应用上下文都是本地线程的,那么区分它们有什么意义呢?
- 使用中间件 DispatcherMiddleware,支持多个 app 共存。就像 request 一样,在多 app 情况下之前也要保证 app 之间的隔离。
- 非 Web 模式下。比如进行测试,一个应用上下文可以有多个请求上下文。但是不能执行 pop 方法,或者使用 with 语句(
__exit__
中会自动执行 pop 方法)。
二、使用上下文
Flask 中有 4 个上下文变量:
- flask.current_app: 应用上下文。它是当前 app 实例对象。
- flask.g: 应用上下文。处理请求时用作临时存储的对象。
- flask.request: 请求上下文。它封装了客户端发出的 HTTP 请求中的内容。
- flask.session: 请求上下文。它存储了用户会话。
其中最常见的就是 flask.g 和 flask.request。
1、使用 flask.request
以下代码段是先引用了 flask.request,但是直到用户访问了 /people/
的时候才通过 request.args.get() 方法获取请求的参数值。试想一下,在引用 flask.request 时,倘若此时还没有用户访问 /people/
,也就是还没有用户发送 /people/
请求,那么这个请求的上下文是怎么获得的呢?
1
2
3
4
5
6
7
from flask import Flask, request
app = Flask(__name__)
@app.route('/people/')
def people():
name = request.args.get('name')
flask.request 就是一个 LocalProxy 实例,这个实例是用来获取名为 _request_ctx_stack 的栈顶对象。以下代码段的逻辑能正常使用,是因为其流程如下:
- 用户访问 /people/ 产生请求
- 在发生请求的过程中向 _request_ctx_stack 推入这个请求上下文的对象,它会变成栈顶。request 就会成为这个请求的上下文,其包含了这次请求相关的信息和数据
- 在视图函数 people() 中使用 request 就可以使用 request.args.get() 获取请求的参数值 name 了
1
2
3
4
5
6
7
8
9
10
11
12
13
14
# coding=utf-8
from functools import partial
from werkzeug.local import LocalStack, LocalProxy
def _lookup_req_object(name):
top = _request_ctx_stack.top
if top is None:
raise RuntimeError('working outside of request context')
return getattr(top, name)
# context locals
_request_ctx_stack = LocalStack()
request = LocalProxy(partial(_lookup_req_object, 'request'))
设想一下,如果不使用 LocalStack 和 LocalProxy 的话,要想让视图函数 people() 访问到请求对象,就只能将其作为参数,一步步传入视图函数中。这样做的缺点是会让每个视图函数都增加一个 request 参数,而 Flask 巧妙地使用上下文把某些对象变为全局可访问(实际上是特定环境的局部对象的代理),每个线程看到的上下文对象确是不同的,这样就巧妙地解决了这个问题。
2、添加上下文的钩子
如下代码中添加了 6 个钩子装饰器,被装饰的函数会注册到 app 中,然后在不同的阶段执行。
- before_first_request: 在处理第一次请求之前执行。
- before_request: 在每次请求前执行。
- teardown_appcontext: 不管是否有异常,注册的函数都会在每次请求之后执行。
- context_processor: 上下文处理的装饰器,返回的字典中的键可以在上下文中使用。
- template_filter: 在使用 Jinja2 模板的时候可以方便地注册过滤器。
- errorhandler: errorhandler 接收状态码,可以自定义返回这种状态码的响应的处理方法。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
# coding=utf-8
import random
from flask import Flask, g, render_template
from ext import db
from users import User
app = Flask(__name__, template_folder='../../templates')
app.config.from_object('config')
db.init_app(app)
def get_current_user():
users = User.query.all()
return random.choice(users)
@app.before_first_request
def setup():
"""
setup() 函数常用来初始化数据,尤其是开发环境下,每次启动应用都会先删掉之前创建的假数据再重新创建
"""
db.drop_all()
db.create_all()
fake_users = [
User('xiaoming', 'xiaoming@dongwm.com'),
User('dongwweiming', 'dongwm@dongwm.com'),
User('admin', 'admin@dongwm.com')
]
db.session.add_all(fake_users)
db.session.commit()
@app.before_request
def before_request():
"""
flask.g 是一个应用上下文,通常放在 before_request 中对它进行数据的填充
"""
g.user = get_current_user()
@app.teardown_appcontext
def teardown(exc=None):
"""
一般来说,对资源的操作有一个 get_X 和一个 teardown_X 对应,多个资源的使用可以使用同一个 teardown 函数。
teardown 通常是做一些环境的清理工作,提交未提交的操作请求等,在本地开发环境和测试时意义较大
"""
if exc is None:
db.session.commit()
else:
db.session.rollback()
db.session.remove()
g.user = None
@app.context_processor
def template_extras():
"""
由于 Jinja2 模板的限制,并不能直接使用 emumerate 这样的 python 自带的函数
(虽然 Jinja2 支持在 for 循环中使用 loop.index 和 loop.index(),但是无法满足全部需要),
可以使用 context_processor 把要用的上下文资源传进去。
这样在模板中就可以直接使用 emumerate 和 current_user 了。
"""
return {'enumerate': enumerate, 'current_user': g.user}
@app.errorhandler(404)
def page_not_found(error):
"""
errorhandler 除了可自定义对不同错误状态码的返回内容,还可以传入自定义的异常对象
"""
return 'This page does not exist', 404
@app.template_filter('capitalize')
def reverse_filter(s):
"""
虽然 Jinja2 支持了非常多的过滤器,但还是无法满足我们的全部需要。
注册一个新的过滤器很方便,以下例子中注册了一个叫作 capitalize 的过滤器,在模板中就可以正常使用(两对花括号和一个管道符)
"""
return s.capitalize()
@app.route('/users')
def user_view():
users = User.query.all()
return render_template('chapter3/section4/user.html', users=users)
if __name__ == '__main__':
app.run(host='0.0.0.0', port=9000)
访问 http://127.0.0.1:9000/users 的结果如下:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
> http get http://127.0.0.1:9000/users
HTTP/1.0 200 OK
Content-Length: 174
Content-Type: text/html; charset=utf-8
Date: Sat, 03 Dec 2022 14:24:55 GMT
Server: Werkzeug/0.11.10 Python/2.7.11+
<h2>Current User: admin</h2>
<ul>
<li>
1 Xiaoming
</li>
<li>
2 Dongwweiming
</li>
<li>
3 Admin
</li>
</ul>
3、使用 LocalProxy 替代 g
现在实现一个全局可访问的 current_user,感受一下 LocalStack 和 LocalProxy 如何工作。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
# coding=utf-8
import random
from flask import Flask, render_template
from werkzeug.local import LocalStack, LocalProxy
from ext import db
from users import User
app = Flask(__name__, template_folder='../../templates')
app.config.from_object('config')
db.init_app(app)
_user_stack = LocalStack()
def get_current_user():
top = _user_stack.top
if top is None:
raise RuntimeError()
return top
current_user = LocalProxy(get_current_user)
@app.before_first_request
def setup():
db.drop_all()
db.create_all()
fake_users = [
User('xiaoming', 'xiaoming@dongwm.com'),
User('dongwweiming', 'dongwm@dongwm.com'),
User('admin', 'admin@dongwm.com')
]
db.session.add_all(fake_users)
db.session.commit()
@app.before_request
def before_request():
users = User.query.all()
user = random.choice(users)
_user_stack.push(user)
@app.teardown_appcontext
def teardown(exc=None):
if exc is None:
db.session.commit()
else:
db.session.rollback()
db.session.remove()
_user_stack.pop()
@app.context_processor
def template_extras():
return {'enumerate': enumerate, 'current_user': current_user}
@app.errorhandler(404)
def page_not_found(error):
return 'This page does not exist', 404
@app.template_filter('capitalize')
def reverse_filter(s):
return s.capitalize()
@app.route('/users')
def user_view():
users = User.query.all()
return render_template('chapter3/section4/user.html', users=users)
if __name__ == '__main__':
app.run(host='0.0.0.0', port=9000)
访问 http://127.0.0.1:9000/users 的结果如下:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
> http get http://127.0.0.1:9000/users
HTTP/1.0 200 OK
Content-Length: 174
Content-Type: text/html; charset=utf-8
Date: Sat, 03 Dec 2022 14:24:55 GMT
Server: Werkzeug/0.11.10 Python/2.7.11+
<h2>Current User: admin</h2>
<ul>
<li>
1 Xiaoming
</li>
<li>
2 Dongwweiming
</li>
<li>
3 Admin
</li>
</ul>