edit
Flask
快速入门
配置与惯例
Flask配置选项的默认值都是合理的，有一些是默认的，如templates和static默认在源码目录下，这些配置可以修改但一般并不需要。
Flask使用了Thread-Local变量，避免了同一request间的参数传递，这样很方便，但是使得程序依赖于合法的request请求，依赖于request的数据导致程序无法复用。
安装
依赖
Flask依赖Werkzeug和Jinja2。
Werkzeug是一个WSGI工具集。
Jinja2负责渲染模板。
virtualenv
sudo pip install virtualenv
mkdir myproject
cd myproject
virtualenv venv
. venv/scripts/activate
pip install Flask
HelloWorld
# 文件hello.py
from flask import Flask
# Flask通过__name__找到包路径及其下的static和templates目录，避免限制为当前目录
app = Flask(__name__)
@app.route('/')
def hello_world():
return 'Hello World!'
if __name__ == '__main__':
app.run()
运行：
python hello.py
通过环境变量运行：
# windows下使用set设置环境变量
export FLASK_APP=hello.py
flask run
通过python module运行：
export FLASK_APP=hello.py
python -m flask run
访问：http://127.0.0.1:5000
定义HttpServer绑定IP
app.run(host='0.0.0.0', port=80)
flask run --host=0.0.0.0
调试模式（交互式调试器）
app.debug = True
app.run()
# 或
app.run(debug=True)
export FLASK_DEBUG=1
flask run
URL路由
基本示例：
@app.route('/')
def index():
return 'Index Page'
@app.route('/hello')
return 'Hello World'
带变量的URL：
@app.route('/user/<username>')
def show_user_profile(username):
return 'User %s' % username
带变量的URL并转换数据类型：
int # 整形
float # 浮点型
string # 不接受斜线
path # 接受斜线的string
any # 任意
uuid # UUID
@app.route('/post/<int:post_id>')
def show_post(post_id):
return 'Post %d' % post_id
遵守唯一URL管理，避免重定向行为:
/projects/ # 若访问/projects会被Flask重定向到/projects/
/about # 若访问/about/会产生"404 Not Found"
建议使用带斜线的URL，保证URL唯一性，避免搜索引擎索引同一个页面两次
使用url_for()用处理函数逆向构造URL：
url_for的第一个参数是处理函数的名称，后面命名参数匹配的部分会应用到url，不符合的部分会作为查询参数。
from flask import Flask, url_for
app = Flask(__name__)
@app.route('/')
def index: pass
@app.route('/login')
def login(username): pass
@app.route('/user/<username>')
def profile(username): pass
# 使用test_request_context()构造request
with app.test_request_context():
print(url_for('index')) # 输出：/
print(url_for('login')) # 输出：/login
print(url_for('login', next='/') # 输出：/login?next=/
print(url_for('profile', username='John Doe')) # 输出：/user/John%20Doe
# url_for默认生成相对地址，外部链接需要使用绝对地址
print(url_or('index', _external=True)) # 输出 http://host:port/
使用逆向构造URL的用途：
- 可以自动在Flask工程中查询URL
- Flask会自动转译特殊字符和Unicode
- 如果应用不位于URL根路径，url_for会自动处理
HTTP方法：
默认处理为GET请求，使用route装饰器的methods指定HTTP方法
@app.route('/login', methods=['GET', 'POST'])
def login():
if request.method == 'POST':
do_the_login()
else:
show_the_login_form()
GET # 获取页面
HEAD # 获取信息，只关心消息头，Flask会自动按GET处理
POST # 发送信息，发送表单
PUT # 类似POST，但可以多次覆盖旧数据，POST只能触发一次
DELETE # 删除
OPTIONS # 查询URL支持哪些HTTP方法，Flask会自动处理
HTTP4和XHTML1只支持GET和POST，但是JavaScript和HTML5支持其他方法
静态文件
文件路径：包或者模块的所在目录中static文件夹
使用URL：/static
生成URL：url_for('static', filename='style.css')，static是特殊参数
模板渲染
模板路径
模块路径：
/application.py
/templates
/hello.html
包路径：
/application
/__init__.py
/templates
/hello.html
渲染：
from flask import render_template
@app.route('/hello/')
@app.route('/hello/<name>')
def hello(name=None)
return render_template('hello.html', name=name)
<!doctype html>
<title>Hello from Flask</title>
{% if name %}
<h1>Hello {{name}}!</h1>
{% else %}
<h1>Hello World!</h1>
{% endif %}
在模板里可以让问request、session和g对象以及get_flashed_message()
模板继承可以使每个页面的header、navigation和footer保持一致。
变量中包含HTML会被转译，例如以上代码{{name}}中包含的HTML会被自动转译。
自动转译是默认对.html .htm .xml .html模板文件开启，模板字符串自动转译是关闭的。
使用|safe过滤器或flask.Markup可以标记字符串为安全。
from flask import Markup
# 标记为安全字符串
Markup('<strong>Hello %s!</strong>') % '<blink>hacker</blink>'
# 输出Markup(u'<strong>Hello <blink>hacker</blink>!</strong>')
# 手动转译
Markup.escape('<blink>hacker</blink>')
# 输出 Markup(u'<blink>hacker</blink>')
# 清除HTML TAG
Markup('<em>Marked up</em> » HTML').striptags()
# 输出 u'Marked up \xbb HTML'
访问Request数据
Request上下文与单元测试
Request使用时看起来是全局的，其实Request是线程（或协程）本地的
单元测试时没有请求，所以没有Request对象
使用test_request_context()环境管理器：
from flask import request
with app.test_request_context('/hello', method='POST')
assert request.path == '/hello'
assert request.method = 'POST'
或将整个WSGI环境传递给request_context()方法
from flask import request
with app.request_context(environ):
assert request.method == 'POST'
使用Reqeust对象
从POST方法的form中获取信息 request.form：
from flask import request
@app.route('/login', methods=['POST', 'GET'])
def login():
error = None
if request.method == 'POST':
if valid_login(request.form['username'], request.form['password']):
return log_the_user_in(request.form['username'])
else:
error = 'Invalid username/password'
return render_template('login.html', error=error)
获取查询字符串 request.args：
# ?key=value
searchword = request.args.get('q', '')
获取json字段：request.args
获取header字段：request.headers
异常处理
若form中属性不存在则会抛出KeyError异常。不处理KeyError异常会显示一个"HTTP 400 Bad Request错误页面”。
上传文件 request.files
from flask import request
@app.route('/upload', methods=['GET', 'POST'])
def upload_file():
if request.method == 'POST':
f = request.files['the_file']
f.save('/var/www/uploads/uploaded_file.txt')
# 获取文件名
f.save('/var/www/uploads/' + secure_filename(f.filename))
上传文件需要HTML表单中设置enctype="multipart/form-data"
Cookies 读取与存储
from flask import request
# 获取cookie
@app.route('/')
def index():
username = request.cookies.get('username')
username = request.cookies.get['username']
# 设置cookie
@app.route('/')
def index():
# 如果不设置cookie只需要返回字符串，若要设置cookie需要自行构造response
# 若不想自行构造response，可以使用“延迟请求回调”
resp = make_response(render_template(...))
resp.set_cookie('username', 'the username')
return resp
重定向与错误
redirect() # 重定向到其他url
from flask import abort, redirect, url_for
@app.route('/')
def index():
return redirect(url_for('login'))
abort() # 发送错误
@app.route('/login')
def login():
abort(401)
this_is_never_executed()
定制错误页面：
from flask import render_template
# 返回模板
@app.errorhandler(404)
def page_not_found(error):
return render_templete('page_not_found.html'), 404
response
url处理函数的返回值会被自动转换为response对象。
如果返回值是字符串，会被转换为字符串为body、状态码为200 OK、MIME是text/html的response。
Flask返回值处理逻辑：
1. 如果返回的是一个合法的response对象，则从url处理函数直接返回。
2. 如果返回的是一个字符串，Flask会用字符串和默认参数创建response。
3. 如果返回一个tuple，格式为(response, status, headers)，Flask会用tuple覆盖默认设置。tuple至少要有一个值。headers是一个list或dict。
4. 如果以上都不是，Flask会假设返回值是一个WSGi应用程序，并发起请求。
如果不想让Flask自动处理，可以使用make_response自行处理。
# 返回tuple
@app.errorhandler(404)
def not_found(error):
return render_template('error.html'), 404
# 自行处理
@app.errorhandler(404)
def not_found(error):
resp = make_response(render_template('error.html'), 404)
resp.headers['X-Something'] = 'A value'
return resp
会话 session
会话是在不同请求间存储特定用户的信息。
Flask使用Cookies字段存储Session，并对Cookies字段做了Hash校验，用户只能查看Cookies内容但不能修改。
from flask import Flask, session, redirect, url_for, escape, request
app = Flask(__name__)
app.secret_key = 'AAAAAAAAAAAAAAAAAAAAAAAAAAAAA'
# 查询session
@app.route('/')
def index():
if 'username' in session:
return 'Logged in as %s' % escape(session['username'])
return 'You are not logged in'
# 插入session
@app.route('/login', methods=['GET', 'POSt'])
def login():
if request.method == 'POST':
session['username'] = request.form['username']
return redirect(url_for('index'))
return '''
<form action="" method="post">
<p><input type=text name=username/></p>
<p><intpu type=submit value=Login></p>
</form>
'''
# 删除session
@app.route('/logout')
def logout():
session.pop('username', None)
return redirect(url_for('index'))
生成密钥：
import os
os.urandom(24)
闪现消息(Flashing)
闪现消息是在请求结束时响应给客户端的一种消息。
当且仅当下一条请求时访问信息。
使用flask()记录消息。
在下一条请求或模版中使用get_flashed_messages()获取消息。
日志记录
app.logger.debug('')
app.logger.warning('', 42)
app.logger.error('')
# logger是标准的python Logger
整合WSGI中间件
使用Werkzeug的中间件避免lighttpd的bugs
from werkzeug.contrib.fixers import LighttpdCGIRootFix
app.wsgi_app = LighttpdCGIRootFix(app.wsgi_app)
使用Flask扩展
见扩展
部署到Web服务器
见部署
简单完整示例教程
示例功能
单用户评论系统
使用sqlite作为数据库
倒叙显示所有评论
评论内容HTTP不安全
工程目录
static存放静态文件
template存放jinja2模板html
/flaskr
/static
/templates
数据库
schema.sql
存放评论的id，标题，内容
drop table if exists entries;
create table entries (
id integer primary key autoincrement,
title string not null,
text string not null
);
创建数据库方法一：使用shell
sqlite3 /tmp/flaskr.db < schema.sql
创建数据库方法二：python
def init_db():
# 在没有请求的时候，flask.g全局变量无法得知应用信息，使用app_context()创建环境。
with app.app_context():
db = get_db()
# open_resource从flask应用目录打开文件。
with app.open_resource('schema.sql', mode='r') as f:
# 读取sql脚本并执行
db.cursor().executescript(f.read())
db.commit()
@app.cli.command('initdb')
def initdb_command():
init_db()
print('Initialized the database.')
Flask应用启动、配置、数据库连接
从环境变量导入路径
Flask没有当前工作目录的概念，因为一个进程中可能会运行多个应用。
可以使用app.root_path获取应用路径，用os.path辅助。
可以使用app.config.from_envvar("FLASKR_SETTINGS", silent=True)从环境变量中获取配置。
使用环境变量传递信息
request是当前请求的环境变量
g是当前request共享信息的变量
数据库连接代码
import os
import sqlite3
from flask import Flask, request, session, g, redirect, url_for, abort, render_template, flash
app = Flask(__name__)
# 默认配置，配置必须使用大写
app.config.update(dict(
DATABASE=os.path.join(app.root_path, 'flaskr.db'),
DEBUG=True,
SECRET_KEY='development key',
USERNAME='admin',
PASSWORD='default'
))
# 从环境变量覆盖配置
app.config.from_envvar('FLASKR_SETTINGS', silent=True)
def connect_db():
'''根据配置文件连接数据库'''
rv = sqlite3.connect(app.config['DATABASE'])
rv.row_factory = sqlite3.Row
return rv
def get_db():
'''保存数据库连接'''
if not hasattr(g, 'sqlite_db'):
g.sqlite_db = connect_db()
return g.sqlite_db
# 使Flask应用结束时销毁的装饰器
@app.teardown_appcontext
def close_db(error):
if hasattr(g, 'sqlite_db'):
g.sqlite_db.close()
if __name__ == '__main__':
app.run()
flask应用，url处理
显示评论
@app.route('/')
def show_entries():
db = get_db()
# 从数据库中读取所有评论
cur = db.execute('select title, text from entries order by id desc')
entries = [dict(title=row[0], text=row[1]) for row in cur.fetchall()]
# 渲染页面
return render_template('show_entries.html', entries=entries)
添加评论
@app.route('/add', methods=['POST'])
def add_entry():
# 在session中查询
if not session.get('logged_in'):
abort(401)
# 写入数据库
db = get_db()
db.execute('insert into entries (title, text) values (?, ?)',
[request.form['title'], request.form['text']])
db.commit()
# 发送falsh消息
flash('New entry was successfully posted')
# 重定向到评论显示页面
return redirect(url_for('show_entries'))
登录页面及登录POST
@app.route('/login', methods=['GET', 'POST'])
def login():
error = None
if request.method == 'POST':
# 登录验证
if request.form['username'] != app.config['USERNAME']:
error = 'Invalid username'
elif request.form['password'] != app.config['PASSWORD']:
error = 'Invalid password'
# 登录成功写入session
else:
session['logged_in'] = True
flash('You were logged in')
return redirect(url_for('show_entries'))
# GET请求返回登录页
return render_template('login.html', error=error)
退出
@app.route('/logout')
def logout():
# 从Session中删除
session.pop('logged_in', None)
flash('You were logged out')
return redirect(url_for('show_entries'))
HTML模板
layout.html
<!doctype html>
<title>Flaskr</title>
<link rel=stylesheet type=text/css href="{{url_for('static', filename='style.css')}}">
<div class=page>
<h1>Flaskr</h1>
<div class=metanav>
{% if not session.logged_in %}
<a href="{{url_for('login')}}">log in</a>
{% else %}
<a href="{{url_for('logout')}}">log out</a>
{% endif %}
</div>
{% for message in get_flashed_messages() %}
<div class=flash>{{ message }}</div>
{% endfor %}
{% block body %}
{% endblock %}
</div>
show_entries.html
使用safe过滤器是jajin2不做安全转换
{% extends "layout.html" %}
{% block body %}
{% if session.logged_in %}
<form action="{{ url_for('add_entry') }}" method=post class=add-entry>
<dl>
<dt>Title:</dt>
<dd><input type=text size=30 name=title></dd>
<dt>Text:</dt>
<dd><textarea name=text rows=5 clos=40></textarea></dd>
<dd><input type=submit value=Share></dd>
<dl>
</form>
{% endif %}
<ul class=entires>
{% for entry in entries %}
<li><h2>{{ entry.title }}</h2>{{ entry.text | safe }}</li>
{% else %}
<li><em>Unbelievable. No entries here so far</em></li>
{% endfor %}
</ul>
{% endblock %}
login.html
{% extends "layout.html" %}
{% block body %}
<h2>Login</h2>
{% if error %}
<p class=error><strong>Error:</strong>{{ error }}</p>
{% endif %}
<form action="" method=post>
<dl>
<dt>Username:</dt>
<dd><input type=text name=username></dd>
<dt>Passwrod:</dt>
<dd><input type=password name=password></dd>
<dd><input type=submit value=Login></dd>
</dl>
</form>
{% endblock %}
CSS样式
style.css
body {font-family: sans-serif; background: #eee;}
a, h1, h2 {color: #377BA8;}
h1, h2 {font-family: 'Georgia', serif; margin:0;}
h1 {border-bottom: 2px solid #eee;}
h2 {font-size: 1.2em;}
.page {margin: 2em auto; width: 35em; border: 5px solid #ccc; padding:0.8em; background: white;}
.entries {list-style: none; margin: 0; padding: 0;}
.entries li {margin: 0.8em 1.2em;}
.entries li h2 {margin-left: -1em;}
.add-entry {font-size: 0.9em; border-bottom: 1px solid #ccc;}
.add-entry dl {font-weight: bold;}
.metanav {text-align: right; font-size: 0.8em; padding: 0.3em; margin-bottom: 1em; background: #fafafa;}
.flash {background: #CEE5F5; padding: 0.5em; border: 1px solid #AACBE2;}
.error {background: #F0D6D6; padding: 0.5em;}
单元测试
测试工程结构
flaskr/
flaskr/
__init__.py
static/
templates/
tests/
test_flaskr.py
setup.py
MANIFEST.in
运行测试
测试的大框架
第一个测试
登录和登出
测试消息的添加
其他测试技巧
伪造资源和上下文
保存上下文
访问和修改Sessions
模板
Jinja配置
默认配置：
所有扩展名为.html .htm .xml .xhtml的木板会开启自动转译
在模板中可以利用{% autoescape false %}标签选择转译是否开启。
Flask在Jinja2上下文中插入了几个全局函数和助手，另外还有一些目前默认的值。
标准上下文，默认在Jinja2模板中可用的值
可以使用的变量和方法
Flask内置变量及方法：
config：flask.config
request：flask.request
session：flask.session
g：flask.g
url_for()：flask.url_for()
get_flasked_message()：flask.get_flashed_message()
python变量：
<p>Dict item: {{ my_dict['key'] }}</p>
<p>List item: {{ my_list[3] }}</p>
<p>List with a variable index: {{ my_list[my_index] }}</p>
<p>Object's method {{ my_object.method() }}</p>
导入宏（Jinja2的宏）的方法
由于以上变量是被添加到request中的，而且不是全局变量，所以想要导入一个需要访问request的宏，需要显式地传入request或变量作为宏的参数。
或者导入宏时手动带上context：
{% from '_helpers.html' import my_macro with context %}
标准过滤器，Jinja2自带的过滤器
tojson():
<script type=text/javascript>
doSomethingWith({{ user.username | tojson | safe }});
</script>
safe：渲染时不转译
capitalize：句子首字母大写
lower：全部小写
upper：全部大写
title：句子中所有词首字母大写
trim：去掉首位空格
striptags：取消所有HTML标签
其他内置过滤器：http://jinja.pocoo.org/docs/templates/#builtin-filters
控制自动转译
Jinja2会自动转译HTML中的特殊字符，如：& > < " '。
但是有时会需要插入HTML，例如插入markdown转换成的HTML，有以下三种方法。
1. 在传递到模板之前，将HTML字符串转换成Markup对象。推荐使用这种方法。
2. 在模版中使用|safe过滤器将变量标记为安全，如{{ var | safe }}。
3. 临时完全禁用自动转译如
{% autoescape false %}
<p>autoescaping is disabled here</p>
<p>{{ will_not_be_escaped_var }}</p>
{% endautoescape %}
注册过滤器
使用装饰器标记自定义Jinja2过滤器
@app.template_filter('reverse')
def reverse_filter(s):
return s[::-1]
手动添加过滤器到Jinja2
def reverse_filter(s):
return s[::-1]
app.jinja_env.filters['reverse'] = reverse_filter
在模板中使用过滤器
{% for x in mylist | reverse %}
{% endfor %}
使用context_processor添加变量、函数添加到模板上下文
context_processor是一个返回dict的函数，dict中的值可以是变量也可以是函数
@app.context_processor
def inject_user():
return dict(user=g.user)
@app.context_processor
def utility_processor():
def format_price(amount, currency=u'$'):
return u'{0:.2f}{1}.format(amount, currency)
return dict(format_price=format_price)
{{ format_price(0.33) }}
模板结构控制
条件
{% if user %}
Hello, {{ user }}!
{% else %}
Hello, Stranger!
{% endif %}
循环
<ul>
{% for comment in comments %}
<li>{{ comment }}</li>
{% endfor %}
</ul>
宏
定义：
{% macro render_comment(comment) %}
<li>{{ comment }}</li>
{% endmacro %}
使用：
<ul>
{% for comment in comments %}
{{ render_comment(comment) }}
{% endfor %}
</ul>
从其他文件中导入：
{% import 'macros.html' as macros %}
<ul>
{% for comment in comments %}
{{ macros.render_comment(comment) }}
{% endfor %}
</ul>
继承
父模板base.html：
<html>
<head>
{% block head %}
<title>{% block title %}{% endblock %} - My Application</title>
{% endblock %}
</head>
<body>
{% block body %}
{% endblock %}
</body>
</html>
子模板：
{% extends "base.html" %}
{% block title %}Index{% endblock %}
{% block head %}
{{ super() }}
<style>
</style>
{% endblock %}
{% block body %}
<h1>Hello, World!</h1>
{% endblock %}
使用super()引用原模板内容，用于向原有块中添加新内容
日志、邮件、调试模式
Flask默认不会写任何日志。
配置Flask发送错误邮件
ADMINS = ['yourname@example.com']
if not app.debug:
import logging
from logging.handlers import SMTPHandler
mail_handler = SMTPHandler('127.0.0.1', 'server-error@example.com', ADMINS, 'YourApplication Failed')
mail_handler.setLevel(logging.ERROR)
app.logger.addHandler(mail_handler)
记录到文件
日志记录器：
- FileHandler 记录文件日志
- RotatingFileHandler 循环覆盖记录日志
- NTEventLogHandler 记录日志到Windows系统事件
- SysLogHandler 记录日志到Unix系统日志
if not app.debug:
import logging
from themodule import TheHandlerYouWant
file_handler = TheHandlerYouWant(...)
file_handler.setLevel(logging.WARNING)
app.logger.addHandler(file_handler)
控制日志格式
邮件格式
from logging import Formatter
mail_handler.setFormatter(Formatter('''
Message type: %(levelname)s
Location: %(pathname)s:%(lineno)d
Module: %(module)s
Function: %(funcName)s
Time: %(asctime)s
Message:
%(message)s
'''))
日志格式
from logging import Formatter
file_handler.setFormatter(Formatter(
'%(asctime)s %(levelname)s: %(message)s '
'[in %(pathname)s:%(lineno)d]'
))
常用日志变量
| 格式 |
描述 |
| %(levelname)s |
消息文本的记录等级 ('DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL'). |
| %(pathname)s |
发起日志记录调用的源文件的完整路径（如果可用） |
| %(filename)s |
路径中的文件名部分 |
| %(module)s |
模块（文件名的名称部分） |
| %(funcName)s |
包含日志调用的函数名 |
| %(lineno)d |
日志记录调用所在的源文件行的行号（如果可用） |
| %(asctime)s |
LogRecord 创建时的人类可读的时间。 |
| %(message)s |
记录的消息，视为 msg % args |
默认情况下，asctime格式为 "2003-07-08 16:49:45,896" （逗号后的数字时间的毫秒部分）。这可以通过继承 :class:~logging.Formatter，并重载 formatTime() 改变。
format(): 处理实际格式，输入LogRecord，输出格式化后的字符串。
formatTime(): 控制asctime格式。
formatException(): 控制异常的格式，输入exc_info，输出字符串。
其他的库
不建议使用Flask.logging统一配置所有日志，因为这样运行多个应用时不能分别配置。
建议使用getLogger获取日志记录，然后遍历日志记录器：
from logging import getLogger
logger = [app.logger, getLogger('sqlalchemy'), getLogger('otherlibrary']
from logger in loggers:
logger.addHandler(mail_handler)
logger.addHandler(file_handler)
开启Debug
app.debug = True
或
run(debug=True)
调试器配置
Flask默认调试器可能会与其他调试器冲突。
- debug: 是否开启调试
- use_debugger: 是否使用内部Flask调试器
- use_reloader: 是否在异常时重新再如并创建子进程
开启外部调试器（Aptana/Eclipse）配置示例：
config.yaml
Flask:
DEBUG: True
DEBUG_WITH_APTANA: True
if __name__ == '__main__':
app = create_app(config='config.yaml')
if app.debug:
# 默认使用Flask内部调试器
use_debugger = True
try:
# 当外部调试器开启时，关闭Flask内部调试器
use_debugger = not(app.config.get('DEBUG_WITH_APTANA'))
except:
pass
app.run(use_debugger=use_debugger, debug=app.debug, use_reloader=use_debugger, host='0.0.0.0')
配置处理
配置基础
内置的配置值
从文件配置
配置的最佳实践
开发/生产
实例文件夹
信号
订阅信号
创建信号
发送信号
信号与Flask的请求上下文
基于装饰器的信号订阅
核心信号
Pluggable Views
基本原则
方法提示
基于调度的方法
装饰试图
基于API的方法视图
应用上下文
应用上下文的作用
在一个Python进程中可以运行多个Flask应用。
为了不使用参数层层传递context，Flask使用current_app指向当前请求所属的应用。
创建应用上下文
当一个request的context被放入处理栈时，如果有需要，应用context就会被一起创建。
显示调用app_context()方法可以创建应用context，如：
from flask import Flask, current_app
app = Flask(__name__)
with app.app_context():
# 在这里就可以通过current_app访问应用context
print current_app.name
print current_app.url_map
应用上下文局部变量
flask.current_app # 程序上下文，当前程序
flask.g # 存储在程序上下文，处理请求时用作临时存储对象，每次请求会重设
flask.request # 请求上下文，请求对象
flask.session # 请求上下文，用户会话
上下文用法
显式创建使用资源
可以在上下文中缓存数据库连接。
import sqlite3
from flask import g
def get_db():
db = getattr(g, '_database', None)
if db is None:
db = g._database = connect_to_database()
return db
@app.teardown_appcontext
def teardown_db(exception):
db = getattr(g, '_database', None)
if db is not None:
db.close()
隐式创建使用资源
使用LocalProxy存储变量
from werkzeug.local import LocalProxy
db = LocalProxy(get_db)
# 用户可以直接通过db访问数据，db已经在内部完成get_db调用
请求上下文
上下文作用域
请求context仅在处理请求时存在。例如以下代码在没有请求时调用会发生错误：
from flask import reqeust, url_for
def redirect_url():
return request.args.get('next') or request.referrer or url_for('index')
可以使用text_request_context('/?next=http://example.com/'模拟请求。
使用with的方式：
with app.text_request_context('/?next=http://example.com/')
redirect_url()
手动入栈请求的方式：
ctx = app.text_request_context('/?next=http://example.com/')
# 入栈
ctx.push()
# 处理请求
redirect_url()
# 出栈
ctx.pop()
Flask请求context内部处理
def wsgi_app(self, environ):
with self.request_context(environ):
try:
response = self.full_dispatch_request()
except Exception, e:
response = self.make_response(self.handle_exception(e))
return response(environ, start_response)
回调和错误
- 在所有请求前调用
before_first_request()。
- 每个请求之前，执行
before_request()上绑定的函数，若某个函数返回了一个response则其他函数将不会被调用。
- 如果
before_request()绑定的函数没有返回response，则url处理函数被正常调用。
- url处理函数返回后被转化成
response对象，之后会调用after_request()
- 最后会调用
teardown_request()上绑定的函数，即使发生异常也会被执行。
teardown_request()
@app.teardown_request
def teardown():
pass
with app.text_client() as client:
resp = client.get('/foo')
# teardown_request()在这里被调用
留意代理
错误时的上下文保护
用蓝图实现模块化
为什么使用蓝图
- 把一个应用分成几个模块。
- 用URL前缀或子域名模块化管理URL。
- 可以使一个模块获得多种URL规则。
- 可以使用蓝图模块化静态文件、模块化模板、模板化过滤器。
- 在Flask扩展中使用。
蓝图的设想
- 注册式地集成到应用。
- 按模块分配请求和处理URL。
创建蓝图（模块）
from flask import Blueprint, render_template, abort
from jinja2 import TemplateNotFound
simple_page = Blueprint('simple_page', __name__, template_folder='templates')
@simple_page.route('/', defaults={'page': 'index'})
@simple_page.route('/<page>')
def show(page):
try:
return render_template('pages/%s.html' % page)
except TemplateNotFound:
abort(404)
蓝图绑定到app时，@simple_page.route会将show注册为simple_page.show
注册（使用）蓝图（模块）
from flask import Flask
from yourapplication.simple_page import simple_page
app = Flask(__name__)
# 挂载在根目录
app.register_blueprint(simple_page)
# 挂载到/pages
app.register_blueprint(simple_page, url_prefix='/pages')
蓝图使用（模块）资源
蓝图资源文件夹
蓝图若使用单独文件夹（Python包），这个文件夹就是蓝图的资源文件夹。
蓝图若共用文件夹（Python模块），模块所在文件夹就是资源文件夹。
可以使用simple_page.root_path获取资源路径。
可以使用simple_page.open_resource()从文件夹打开文件。如：
with simple_page.open_resource('static/style.css') as f:
code = f.read()
静态文件
可以使用static_folder参数指定静态文件夹路径，如：
admin = Blueprint('admin', __name__, static_folder='static')
若admin模块加载在根路径'/'，静态资源路径就是'/static/xxx'
若admin模块加载在'/admin'，静态资源路径就是'/admin/static/xxx'
可以使用admin.static访问static参数，如：
url_for('admin.static', filename='style.css')
模板路径
admin = Blueprint('admin', __name__, template_folder='templates')
构造URL
# 不同模块构造url
url_for('admin.index')
# 相同模块构造url
url_for('.index')
Flask扩展
寻找扩展
使用扩展
Flask 0.8以前
Flask-Script
命令交互式操作Flask
from flask import Flask
# 废弃
from flask.ext.script import Manager
# 建议
from flask_script import Manager
app = Flask(__name__)
manager = Manager(app)
if __name__ == '__main__':
manager.run()
py -3 hello.py --help
py -3 hello.py shell --help
py -3 hello.py runserver --help
自动导入上下文：
from flask_script import Shell
def make_shell_context():
return dict(app=app, db=db, User=User, Role=Role)
manager.add_command('shell', Shell(make_context=make_shell_context))
Flask-Bootstrap
flask_bootstrap提供bootstrap，google analytics，WTForm定义等模板
base.html定义的块
| 块名 |
说明 |
| doc |
整个HTML文档 |
| html_attribs |
标签的属性 |
| html |
标签中的内容 |
| head |
标签中的内容 |
| title |
标签中的内容</td>
</tr>
<tr>
<td>metas</td>
<td>一组<meta>标签</td>
</tr>
<tr>
<td>styles</td>
<td>层叠样式表定义</td>
</tr>
<tr>
<td>body_attribs</td>
<td><body>标签的属性</td>
</tr>
<tr>
<td>body</td>
<td><body>标签中的内容</td>
</tr>
<tr>
<td>navbar</td>
<td>用户定义的导航条</td>
</tr>
<tr>
<td>content</td>
<td>用户定义的页面内容</td>
</tr>
<tr>
<td>scripts</td>
<td>文档底部的Javascript声明</td>
</tr>
</tbody>
</table>
<p>避免覆盖模板原有内容：</p>
<pre class="codehilite"><code>{% block scripts %}
{{ super() }}
<script type="text/javascript" src="my-script.js"></script>
{% endblock %}</code></pre>
<h4 id="python">python初始化<a class="headerlink" href="#python" title="Permanent link">¶</a></h4>
<pre class="codehilite"><code class="language-py">from flask import app
from flask_bootstrap import Bootstrap
app = Flask(__name__)
bootstrap = Bootstrap(app)
app.run()</code></pre>
<h4 id="_90">模板继承<a class="headerlink" href="#_90" title="Permanent link">¶</a></h4>
<pre class="codehilite"><code class="language-py">{% extends "bootstrap/base.html" %}
{% block title %}Flasky{% endblock %}
{% block navbar %}
<div class="navbar navbar-inverse" role="navigation">
<div class="container">
<div class="navbar-header">
<button type="button" class="navbar-toggle" data-toggle="collapse" data-target=".navbar-collapse">
<span class="sr-only">Toggle navigation</span>
<span class="icon-bar"></span>
<span class="icon-bar"></span>
<span class="icon-bar"></span>
</button>
<a class="navbar-brand" href="/">Flasky</a>
</div>
<div class="navbar-collapse collapse">
<ul class="nav navbar-nav">
<li><a href="/">Home</a></li>
</ul>
</div>
</div>
</div>
{% endblock %}
{% block content %}
<div class="container">
<div class="page-header">
<h1>Hello, {{ name }}!</h1>
</div>
</div>
{% endblock %}</code></pre>
<h3 id="flask-moment">Flask-Moment<a class="headerlink" href="#flask-moment" title="Permanent link">¶</a></h3>
<p>moment.js提供本地化时间<br />
http://momentjs.com/docs/#/displaying/</p>
<p>安装：</p>
<pre class="codehilite"><code class="language-sh">pip install flask-moment</code></pre>
<p>python：</p>
<pre class="codehilite"><code class="language-py">from datetime import datetime
from flask.ext.moment import Moment
moment = Moment(app)
@app.route('/')
def index():
return render_template('index.html', current_time=datetime.utcnow())</code></pre>
<p><strong>moment需要输入不带时区的utc时间</strong></p>
<p>模板：</p>
<pre class="codehilite"><code>{% block scripts %}
{{ super() }}
{{ moment.include_moment() }}
{% endblock %}
{% block page_content %}
<p>The local date and time is {{ moment(current_time).format('LLL') }}.</p>
<p>That was {{ moment(current_time).fromNow(refresh=True) }}.</p>
{% endblock %}</code></pre>
<p><strong>moment.format显示时间</strong><br />
<strong>moment.fromNow显示距离当前多久，动态刷新</strong> </p>
<p>语言本地化：</p>
<pre class="codehilite"><code class="language-py">{{ moment.lang('es') }}
{{ moment.lang('zh-cn') }}</code></pre>
<h3 id="flask-wtf">Flask-WTF<a class="headerlink" href="#flask-wtf" title="Permanent link">¶</a></h3>
<p>flask-wtf默认开启防止csrf攻击，即发送表单时发送一个key要求响应时带回，key错误时为伪造响应。</p>
<p>安装：</p>
<pre class="codehilite"><code class="language-sh">pip install flask-wtf</code></pre>
<p>启动:</p>
<pre class="codehilite"><code class="language-py">app = Flask(__name__)
app.config['SECRET_KEY'] = 'flask-wtf需要使用secret_key'</code></pre>
<p>定义：</p>
<pre class="codehilite"><code class="language-py">from flask_wtf import Form
from wtforms import StringField, SubmitField
from wtforms.validators import Required
class NameForm(Form):
name = StringField('What is your name?', validators=[Required()])
submit = SubmitField('Submit')</code></pre>
<p>wtf手动生成表单：</p>
<pre class="codehilite"><code><form method="POST">
{{ form.hidden_tag() }}
{{ form.name.label }} {{ form.name() }}
{{ form.name.label }} {{ form.name(id='my-text-field') }} # 指定ID等用于样式表
{{ form.submit() }}
</form></code></pre>
<p>flask-bootstrap + flask-wtf自动生成表单：</p>
<pre class="codehilite"><code>{% import "bootstrap/wtf.html" as wtf %}
{{ wtf.quick_form(form) }}</code></pre>
<p>表单验证：</p>
<pre class="codehilite"><code class="language-py">@app.route('/', methods=['GET', 'POST'])
def index():
name = None
form = NameForm()
if form.validate_on_submit():
name = form.name.data
form.name.date = ''
return render_template('index.html', form=form, name=name)</code></pre>
<p><strong>GET POST二合一，POST时通过validate_on_submit判断获取输入</strong></p>
<p>POST后刷新处理：<br />
POST后刷新，浏览器会重复POST，可以通过POST响应URL重定向避免。 </p>
<pre class="codehilite"><code class="language-py">from flask import Flask, render_template, session, redirect, url_for
@app.route('/', methods=['GET', 'POST'])
def index():
form = NameForm()
if form.validate_on_submit():
session['name'] = form.name.data
return redirect(url_for('index'))
return render_template('index.html', form=form, name=session.get('name'))</code></pre>
<p>字段：<br />
字段类型 | 说明<br />
---------|-----<br />
StringField | 文本字段<br />
TextAreaField | 多行文本字段<br />
PasswordField | 密码文本字段<br />
HiddenField | 隐藏文本字段<br />
DateField | 文本字段，值为datetime.date格式<br />
DateTimeField | 文本字段，值为datetime.datetime格式<br />
IntegerField | 文本字段，值为整数<br />
DecimalField | 文本字段，值为decimal.Decimal<br />
FloatField | 文本字段，值为浮点数<br />
BooleanField | 复选框，值为True和False<br />
RadioField | 一组单选框<br />
SelectField | 下拉列表<br />
SelectMultipliField | 下拉列表，可选择多个值<br />
FileField | 文件上传字段<br />
SubmitField | 表单提交按钮<br />
FormField | 把表单作为字段嵌入另一个表单<br />
FieldList | 一组指定类型的字段</p>
<p>验证：<br />
验证函数 | 说明<br />
---------|-----<br />
Email | 验证电子邮件地址 <br />
EqualTo | 比较两个字段的值，常用于要求输入两次密码进行确认<br />
IPAddress | 验证IPv4网络地址<br />
Length | 验证输入字符串的长度<br />
NumberRange | 验证输入的值在数字范围内<br />
Optional | 无输入时跳过其他验证函数 <br />
Required | 确保字段中有数据 <br />
Regexp | 使用正则表达式验证输入值<br />
URL | 验证URL<br />
AnyOf | 确保输入值在列表中<br />
NoneOf | 确保输入值不在列表中 </p>
<h3 id="flask-sqlalchemy">Flask-SQLAlchemy<a class="headerlink" href="#flask-sqlalchemy" title="Permanent link">¶</a></h3>
<p>安装：</p>
<pre class="codehilite"><code class="language-sh">pip install flask-sqlalchemy</code></pre>
<p>数据库URI配置：</p>
<pre class="codehilite"><code class="language-sh">export SQLALCHEMY_DATABASE_URI=mysql://username:password@hostname/database</code></pre>
<table>
<thead>
<tr>
<th>数据库引擎</th>
<th>URL</th>
</tr>
</thead>
<tbody>
<tr>
<td>MySQL</td>
<td>mysql://username:password@hostnmae/database</td>
</tr>
<tr>
<td>Postgres</td>
<td>postgresql://username:password@hostname/database</td>
</tr>
<tr>
<td>SQLite(*nix)</td>
<td>sqlite:///absolute/path/to/database</td>
</tr>
<tr>
<td>SQLite(win)</td>
<td>sqlite:///c:/absolute/path/to/database</td>
</tr>
</tbody>
</table>
<p>初始化:</p>
<pre class="codehilite"><code class="language-py">from flask_salalchemy import SQLAlchemy
basedir = os.path.abspath(os.path.dirname(__file__))
app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///' + os.path.join(basedir, 'data.sqlite')
# 自动提交
app.config['SQLALCHEMY_COMMIT_ON_TEARDOWN'] = True
# 跟踪数据库变更
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
db = SQLAlchemy(app)</code></pre>
<p>模型定义：</p>
<pre class="codehilite"><code class="language-py">class Role(db.Model):
__tablename__ = 'roles'
id = db.Column(db.Integer, primary_key=True)
name = db.Column(db.String(64), unique=True)
def __repr__(self):
return '<Role %r>' % self.name
class User(db.Model):
__tablename__ = 'users'
id = db.Column(db.Integer, primary_key=True)
username = db.Column(db.String(64), unique=True, index=True)
def __repr__(self):
return '<User %r>' % self.username</code></pre>
<p><strong>__repr()__返回一个可读性字符串表示模型，用于调试</strong></p>
<p>操作：</p>
<pre class="codehilite"><code class="language-py">from app.py import db, Role, User
# 不存在则创建数据库
db.create_all()
# 删除
db.drop_all()
# 创建对象
admin_role = Role(name='Admin')
mod_role = Role(name='Moderator')
user_role=Role(name='User')
user_john=User(username='john', role=admin_role)
user_susan = User(username='susan', role=user_role)
user_david = User(username='david', role=user_role)
# 插入数据库
db.session.add(admin_role)
db.session.add(mod_role)
db.session.add(user_role)
db.session.add(user_john)
db.session.add(user_susan)
db.session.add(user_david)
# 批量插入数据库
db.session.add_all([admin_role, mod_role, user_role, user_john, user_susan, user_david])
# 提交，session是原子操作，避免不一致性
db.session.commit()
# 回退
db.session.rollback()
# 修改
admin_role.name = 'Administrator'
db.session.add(admin_role)
db.session.commit()
# 删除
db.session.delete(mod_role)
db.session.commit()
# 查询
Role.query.all()
User.query.all()
User.query.filter_by(role=user_role)
Role.query.filter_by(name='User').first()
# 显示SQL语句
str(User.query.filter_by(role=user_role))
# 排序
user_role.users.order_by(User.username).all()
# 统计
user_role.users.count()</code></pre>
<p>数据类型：<br />
类型名 | Python类型 | 说明<br />
-------|------------|------<br />
Integer | int | 普通整数，一般是32位<br />
SmallInteger | int | 取值范围小的整数，一般是16位<br />
BigInteger | int或long | 不限制精度的整数<br />
Float | float | 浮点数<br />
Numeric | decimal.Decimal | 定点数<br />
String | str | 变长字符串<br />
Text | str | 变长字符串，对较长或不限长度的字符串做了优化<br />
Unicode | unicode | 变长Unicode字符串<br />
UnicodeText | unicode | 变长Unicode字符串，对较长或不限长度的字符串做了优化<br />
Boolean | bool | 布尔值<br />
Date | datetime.date | 日期<br />
Time | datetime.time | 时间<br />
DateTime | datetime.datetime | 日期和世界<br />
Interval | datetime.timedelta | 时间间隔<br />
Enum | str | 一组字符串<br />
PickleType | 任何Python对象 | 自动使用Pickle序列化<br />
LargeBinary | str | 二进制文件</p>
<p>列选项：<br />
选项名 | 说明<br />
-------|-----<br />
primary_key | 主键<br />
unique | 不允许重复<br />
index | 创建索引<br />
nullable | 允许使用空值<br />
default | 定义默认值</p>
<p>一对多关系：</p>
<pre class="codehilite"><code class="language-py">class Role(db.Model):
users = db.relationship('User', backref='role')
class User(db.Model):
role_id = db.Column(db.Integer, db.ForeignKey('roles.id'))</code></pre>
<p><strong>db.relationship定义一种特殊属性users，访问users属性返回对应的User列表</strong><br />
<strong>backref在User模型中添加一个role属性，访问role访问的是对象，访问role_id只能访问外键的值</strong> </p>
<p>relationship()选项：<br />
选项名 | 说明<br />
-------|-----<br />
backref | 在关系的另一个模型中添加反向作用<br />
primaryjoin | 明确指定两个模型之间使用的联结条件。只在模棱两可的关系中需要定义。<br />
lazy | 加载方式:<br />
... | select：首次访问时按需加载<br />
... | immediate：源对象加载则加载<br />
... | join：使用联结加载<br />
... | subquery：立即加载，使用子查询<br />
... | noload：永不加载<br />
... | dynamic：不加载，提供查询<br />
uselist | True使用列表，False一对一不使用列表<br />
order_by | 指定关系中记录的排序方式<br />
secondary | 指定多对多关系中关系表的名称<br />
secondaryjoin | SQLAlchemy无法决定时，指定多对多关系中的二级联结条件</p>
<p>查询过滤器：<br />
过滤器 | 说明<br />
-------|-----<br />
filter() | 把过滤器添加到原查询上，返回一个新查询<br />
filter_by() | 把等值过滤器添加到原查询上，返回一个新查询<br />
limit() | 限制查询结果数量，返回一个新查询<br />
offset() | 便宜查询结果，返回一个新查询<br />
order_by() | 排序，返回新查询<br />
group_by() | 分组，返回新查询</p>
<p>查询执行函数：<br />
方法 | 说明<br />
-----|-----<br />
all() | 返回所有结果list<br />
first() | 返回第一个结果或None<br />
first_or_404() | 返回第一个结果或404<br />
get() | 返回指定主键对应行或None<br />
get_or_404() | 返回指定主键对应行或404<br />
count() | 返回查询结果的数量<br />
paginate() | 返回一个Paginate对象，包含范围内的结果</p>
<h3 id="_91">数据库迁移<a class="headerlink" href="#_91" title="Permanent link">¶</a></h3>
<p>SQLAlchemy Alembic:<a href="https://alembic.readthedocs.org/en/latest/index.html">https://alembic.readthedocs.org/en/latest/index.html</a><br />
Flask-Migrate(Alembic轻量级包装):<a href="http://flask-migrate.readthedocs.org/en/latest/">http://flask-migrate.readthedocs.org/en/latest/</a>，依赖Flask-Script操作。</p>
<p>安装：</p>
<pre class="codehilite"><code class="language-sh">pip install flask-migrate</code></pre>
<p>启动代码：</p>
<pre class="codehilite"><code class="language-py">from flask_migrate import Migrate, MigrateCommand
app = Flask(__name__)
db = SQLAlchemy(app)
migrate = Migrate(app, db)
manager.add_command('db', MigrateCommand)</code></pre>
<p>初始化命令：</p>
<pre class="codehilite"><code class="language-sh">python hello.py db init</code></pre>
<p>手动创建迁移（自行实现<code>upgrade()</code>和<code>downgrade()</code>）：</p>
<pre class="codehilite"><code class="language-sh">python hello.py db revision</code></pre>
<p>自动创建迁移（需要检查）</p>
<pre class="codehilite"><code class="language-sh">python hello.py db migrate -m "initial migration"</code></pre>
<p>升级数据库</p>
<pre class="codehilite"><code class="language-sh">python hello.py db upgrade</code></pre>
<h3 id="flask-mail">Flask-Mail<a class="headerlink" href="#flask-mail" title="Permanent link">¶</a></h3>
<p>包装python标准smtplib</p>
<p>安装：</p>
<pre class="codehilite"><code class="language-sh">pip install flask-mail</code></pre>
<p>配置项：<br />
配置 | 默认值 | 说明<br />
-----|--------|-----<br />
MAIL_SERVER | localhost | 电子邮件服务器域名或IP<br />
MAIL_PORT | 25 | 电子邮件服务器端口<br />
MAIL_USE_TLS | False | 启用TLS<br />
MAIL_USE_SSL | False | 启用SSL<br />
MAIL_USERNAME | None | 邮件账户的用户名<br />
MAIL_PASSWORD | None | 邮件账户的密码</p>
<p>配置启动：</p>
<pre class="codehilite"><code class="language-py">import os
from flask import Flask
from flask_mail import Mail
app = Flask(__name__)
app.config['MAIL_SERVER'] = 'smtp.googlemail.com'
app.config['MAIL_PORT'] = 587
app.config['MAIL_USE_TLS'] = True
app.config['MAIL_USERNAME'] = os.evviron.get('MAIL_USERNAME')
app.config['MAIL_PASSWORD'] = os.environ.get('MAIL_PASSWORD')
mail = Mail(app)</code></pre>
<p>发送邮件：</p>
<pre class="codehilite"><code class="language-py">from flask.ext.mail import Message
msg = Message('test subject', sender='you@example.com', recipients=['you@example.com'])
msg.body = 'text body'
msg.html = '<b>HTML</b> body'
mail.send(msg)</code></pre>
<p>发送邮件（模板渲染）：</p>
<pre class="codehilite"><code class="language-py">from flask.ext.mail import Message
msg = Message('test subject', sender='you@example.com', recipients=['you@example.com'])
msg.body = render_template(template + '.txt', parm1, parm2)
msg.html = render_template(template + '.html', parm1, parm2)
mail.send(msg)</code></pre>
<h3 id="celery">Celery任务队列<a class="headerlink" href="#celery" title="Permanent link">¶</a></h3>
<h4 id="_92">特性<a class="headerlink" href="#_92" title="Permanent link">¶</a></h4>
<p>Brokers(消息丢列，代理): RabbitMQ, Redis, Amazon SQS<br />
Result Stores(结果存储): AMQP, Redis, Memcached, SQLAlchemy, Django ORM, Apache Cassandra, Elasticsearch<br />
Concurrency(并发): prefork(multiprocessing), Eventlet, gevent, solo(single threaded)<br />
Serialization(序列化): pickle, json, yaml, msgpack. zlib, bzip2, 消息签名</p>
<p>Monitoring: worker实时监控事件<br />
Work-flows: 工作流支持，分组、数据链、块<br />
Time & Rate Limits: 按时间、处理时长进行流控<br />
Scheduling: 计划任务，相对时间、时间间隔、定时任务 </p>
<h4 id="_93">安装<a class="headerlink" href="#_93" title="Permanent link">¶</a></h4>
<pre class="codehilite"><code class="language-sh">pip install celery
pip install celery[redis]
# 其他Broker
pip install celery[librabbitmq, redis, auth, msgpack]</code></pre>
<h4 id="_94">定义任务<a class="headerlink" href="#_94" title="Permanent link">¶</a></h4>
<p>task.py</p>
<pre class="codehilite"><code class="language-py">from celery import Celery
# redis url 格式:'redis://:password@hostname:port/db_number'
# 默认port为6379，数据库id为0
# redis socket 格式: 'redis+socket:///path/to/redis.sock?virtual_host=db_number'
# 只执行不返回结果
# app = Celery('tasks', broker='redis://localhost')
# 配置backend保存结果
app = Celery('tasks', backend='redis://localhost', broker='redis://localhost')
@app.task()
def add(x, y)
return x + y</code></pre>
<h4 id="worker">启动Worker任务<a class="headerlink" href="#worker" title="Permanent link">¶</a></h4>
<pre class="codehilite"><code class="language-sh">celery -A task worker # 启动任务
celery -A task worker --loglevel=info # 带日志选项启动任务
celery worker --help # 查看worker选项帮助
celery help # 帮助</code></pre>
<h4 id="_95">发送任务<a class="headerlink" href="#_95" title="Permanent link">¶</a></h4>
<p>client.py</p>
<pre class="codehilite"><code class="language-py">from task import add
result = add.delay(1, 2) # 返回一个AsyncResult对象，用于检查状态、获取结果或异常
result = add.apply_async((1, 2))
# 检查状态
result.ready()
# 获取结果
result.get() # 必须设置backend保存及诶过
# 关闭异常重现，默认若出现异常get会重新抛出异常
result.get(propagate=False)
# 访问异常调用栈
result.traceback</code></pre>
<h4 id="celery-flower">Celery Flower<a class="headerlink" href="#celery-flower" title="Permanent link">¶</a></h4>
<p><a href="https://flower.readthedocs.io/en/latest/">https://flower.readthedocs.io/en/latest/</a></p>
<p>特性：<br />
- 实时任务执行监控工具web界面<br />
- 远程控制<br />
- Broker监控<br />
- HTTP API<br />
- 基本用户认证，支持Google OpenID</p>
<pre class="codehilite"><code class="language-sh">pip install flower
flower -A proj --port=5555
curl http://localhost:5555/</code></pre>
<h3 id="flask-admin">Flask-Admin<a class="headerlink" href="#flask-admin" title="Permanent link">¶</a></h3>
<h2 id="shell">与Shell共舞<a class="headerlink" href="#shell" title="Permanent link">¶</a></h2>
<h3 id="_96">创建一个请求上下文<a class="headerlink" href="#_96" title="Permanent link">¶</a></h3>
<h3 id="_97">激发请求发送前后的调用<a class="headerlink" href="#_97" title="Permanent link">¶</a></h3>
<h2 id="flask_5">Flask代码模式<a class="headerlink" href="#flask_5" title="Permanent link">¶</a></h2>
<h3 id="url_2">大型应用（批量导入url处理类)<a class="headerlink" href="#url_2" title="Permanent link">¶</a></h3>
<h4 id="_98">修改前目录结构<a class="headerlink" href="#_98" title="Permanent link">¶</a></h4>
<pre class="codehilite"><code>/application
/application.py
/static
/style.css
/templates
/layout.html
/index.html
/login.html
...</code></pre>
<h4 id="_99">修改后目录结构<a class="headerlink" href="#_99" title="Permanent link">¶</a></h4>
<pre class="codehilite"><code>/application
/runserver.py
/application
/__init__.py
/views.py
/static
/templates</code></pre>
<p>添加runserver.py</p>
<pre class="codehilite"><code class="language-py">from application import app
app.run(debug=True)</code></pre>
<p>添加<strong>init</strong>.py</p>
<pre class="codehilite"><code class="language-py">from flask import Flask
app = Flask(__name__)
# 在Flask生成后导入模块中所有view
import application.views</code></pre>
<p>添加views.py</p>
<pre class="codehilite"><code class="language-py">from application import app
@app.route('/')
def index():
return 'Hello World!'</code></pre>
<p>修改步骤：<br />
1. 使用python包（目录）替换python模块（py文件）管理代码<br />
2. 添加启动代码runserver.py<br />
3. 在包<strong>init</strong>.py中创建Flask应用，导入模块所有view。<br />
4. 在views中引用app，实现url处理程序。 </p>
<p><strong>这样修改有循环依赖的嫌疑，其实在__init__.py中只导入并没有使用，并且实在文件末尾导入的</strong></p>
<h3 id="_100">应用程序的工厂函数<a class="headerlink" href="#_100" title="Permanent link">¶</a></h3>
<h3 id="_101">应用调度<a class="headerlink" href="#_101" title="Permanent link">¶</a></h3>
<h3 id="api_1">实现API异常处理<a class="headerlink" href="#api_1" title="Permanent link">¶</a></h3>
<h3 id="url_3">使用URL处理器<a class="headerlink" href="#url_3" title="Permanent link">¶</a></h3>
<h3 id="setuptools">使用Setuptools部署、分发<a class="headerlink" href="#setuptools" title="Permanent link">¶</a></h3>
<h3 id="fabric">使用Fabric部署<a class="headerlink" href="#fabric" title="Permanent link">¶</a></h3>
<h3 id="flasksqlite3">在Flask中使用sqlite3<a class="headerlink" href="#flasksqlite3" title="Permanent link">¶</a></h3>
<h3 id="flasksqlalchemy">在Flask中使用sqlalchemy<a class="headerlink" href="#flasksqlalchemy" title="Permanent link">¶</a></h3>
<h3 id="_102">上传文件<a class="headerlink" href="#_102" title="Permanent link">¶</a></h3>
<h3 id="_103">缓存<a class="headerlink" href="#_103" title="Permanent link">¶</a></h3>
<h3 id="_104">视图装饰器<a class="headerlink" href="#_104" title="Permanent link">¶</a></h3>
<h3 id="wtforms">使用WTForms进行表单验证<a class="headerlink" href="#wtforms" title="Permanent link">¶</a></h3>
<h3 id="_105">模板继承<a class="headerlink" href="#_105" title="Permanent link">¶</a></h3>
<h3 id="_106">消息闪现<a class="headerlink" href="#_106" title="Permanent link">¶</a></h3>
<h3 id="jqueryajax">用jQuery实现Ajax<a class="headerlink" href="#jqueryajax" title="Permanent link">¶</a></h3>
<h3 id="_107">自定义错误页面<a class="headerlink" href="#_107" title="Permanent link">¶</a></h3>
<h3 id="_108">延迟加载视图<a class="headerlink" href="#_108" title="Permanent link">¶</a></h3>
<h3 id="flaskmongokit">在Flask中使用MongoKit<a class="headerlink" href="#flaskmongokit" title="Permanent link">¶</a></h3>
<h3 id="favicon">添加Favicon<a class="headerlink" href="#favicon" title="Permanent link">¶</a></h3>
<h4 id="_109">特性<a class="headerlink" href="#_109" title="Permanent link">¶</a></h4>
<p>16*16像素的ICO文件</p>
<h4 id="_110">新标准<a class="headerlink" href="#_110" title="Permanent link">¶</a></h4>
<pre class="codehilite"><code class="language-html"><link rel="shortcut icon" href="{{ url_for('static', filename='favicon.ico') }}"></code></pre>
<h4 id="_111">旧标准<a class="headerlink" href="#_111" title="Permanent link">¶</a></h4>
<p>放在root路径下的favicon.ico</p>
<pre class="codehilite"><code class="language-py">app.add_url_rule('/favicon.ico', redirect_to=url_for('static', filename='favicon.ico'))</code></pre>
<p>或者</p>
<pre class="codehilite"><code class="language-py">import os
from flask import send_from_directory
@app.route('/favicon.ico')
def favicon():
return send_from_directory(os.path.join(app.root_path, 'static'), 'favicon,ico', mimetype='image/vnd.microsoft.icon')</code></pre>
<h4 id="_112">模板<a class="headerlink" href="#_112" title="Permanent link">¶</a></h4>
<p>templates/base.html</p>
<pre class="codehilite"><code>{% block head %}
{{ super() }}
<link rel="shortcut icon" href="{{ url_for('static', filename='favicon.ico') }}" type="image/x-icon">
<link rel="icon" href="{{ url_for('static', filename='favicon.ico') }} type="image/x-icon">
{% endblock %}</code></pre>
<h3 id="_113">数据流<a class="headerlink" href="#_113" title="Permanent link">¶</a></h3>
<h3 id="_114">延迟请求回调<a class="headerlink" href="#_114" title="Permanent link">¶</a></h3>
<h3 id="http-method-overrides">添加HTTP Method Overrides<a class="headerlink" href="#http-method-overrides" title="Permanent link">¶</a></h3>
<h3 id="_115">请求内容校验码<a class="headerlink" href="#_115" title="Permanent link">¶</a></h3>
<h3 id="celery_1">基于Celery的后台任务<a class="headerlink" href="#celery_1" title="Permanent link">¶</a></h3>
<h3 id="flask_6">继承Flask<a class="headerlink" href="#flask_6" title="Permanent link">¶</a></h3>
<h2 id="_116">部署选择<a class="headerlink" href="#_116" title="Permanent link">¶</a></h2>
<h3 id="_117">托管配置<a class="headerlink" href="#_117" title="Permanent link">¶</a></h3>
<h3 id="_118">主机配置<a class="headerlink" href="#_118" title="Permanent link">¶</a></h3>
<h4 id="mod_wsgiapache">mod_wsgi（Apache）<a class="headerlink" href="#mod_wsgiapache" title="Permanent link">¶</a></h4>
<h4 id="wsgi_1">独立WSGI容器<a class="headerlink" href="#wsgi_1" title="Permanent link">¶</a></h4>
<p>Gunicorn: </p>
<p>Gevent: </p>
<p>Twisted Web: </p>
<p>Proxy Setups: </p>
<h4 id="uwsgi">uWSGI<a class="headerlink" href="#uwsgi" title="Permanent link">¶</a></h4>
<h4 id="fastcgi">FastCGI<a class="headerlink" href="#fastcgi" title="Permanent link">¶</a></h4>
<h4 id="cgi">CGI<a class="headerlink" href="#cgi" title="Permanent link">¶</a></h4>
<h2 id="_119">大型工程<a class="headerlink" href="#_119" title="Permanent link">¶</a></h2>
<h3 id="_120">阅读源码<a class="headerlink" href="#_120" title="Permanent link">¶</a></h3>
<h3 id="_121">钩子，继承<a class="headerlink" href="#_121" title="Permanent link">¶</a></h3>
<h3 id="_122">继承<a class="headerlink" href="#_122" title="Permanent link">¶</a></h3>
<h3 id="_123">用中间件包装<a class="headerlink" href="#_123" title="Permanent link">¶</a></h3>
<h3 id="_124">分支<a class="headerlink" href="#_124" title="Permanent link">¶</a></h3>
<h3 id="_125">像专家一样扩大规模<a class="headerlink" href="#_125" title="Permanent link">¶</a></h3>
<h3 id="_126">与社区对话<a class="headerlink" href="#_126" title="Permanent link">¶</a></h3>
<h2 id="api_2">API<a class="headerlink" href="#api_2" title="Permanent link">¶</a></h2>
<h3 id="_127">应用对象<a class="headerlink" href="#_127" title="Permanent link">¶</a></h3>
<h3 id="_128">蓝图对象<a class="headerlink" href="#_128" title="Permanent link">¶</a></h3>
<h3 id="_129">进入的请求对象<a class="headerlink" href="#_129" title="Permanent link">¶</a></h3>
<h3 id="_130">响应对象<a class="headerlink" href="#_130" title="Permanent link">¶</a></h3>
<h3 id="_131">会话<a class="headerlink" href="#_131" title="Permanent link">¶</a></h3>
<h3 id="_132">会话接口<a class="headerlink" href="#_132" title="Permanent link">¶</a></h3>
<h3 id="_133">测试客户端<a class="headerlink" href="#_133" title="Permanent link">¶</a></h3>
<h3 id="_134">应用全局变量<a class="headerlink" href="#_134" title="Permanent link">¶</a></h3>
<h3 id="_135">有用的函数和类<a class="headerlink" href="#_135" title="Permanent link">¶</a></h3>
<h3 id="_136">消息闪现<a class="headerlink" href="#_136" title="Permanent link">¶</a></h3>
<h3 id="json">JSON支持<a class="headerlink" href="#json" title="Permanent link">¶</a></h3>
<h3 id="_137">模板渲染<a class="headerlink" href="#_137" title="Permanent link">¶</a></h3>
<h3 id="_138">配置<a class="headerlink" href="#_138" title="Permanent link">¶</a></h3>
<h3 id="_139">扩展<a class="headerlink" href="#_139" title="Permanent link">¶</a></h3>
<h3 id="_140">流的辅助函数<a class="headerlink" href="#_140" title="Permanent link">¶</a></h3>
<h3 id="_141">有用的内构件<a class="headerlink" href="#_141" title="Permanent link">¶</a></h3>
<h3 id="_142">信号<a class="headerlink" href="#_142" title="Permanent link">¶</a></h3>
<h3 id="_143">基于类的视图<a class="headerlink" href="#_143" title="Permanent link">¶</a></h3>
<h3 id="url_4">URL路由注册<a class="headerlink" href="#url_4" title="Permanent link">¶</a></h3>
<h3 id="_144">视图函数选项<a class="headerlink" href="#_144" title="Permanent link">¶</a></h3>
<h2 id="_145">其他说明<a class="headerlink" href="#_145" title="Permanent link">¶</a></h2>
<h3 id="flask_7">Flask中的设计决策<a class="headerlink" href="#flask_7" title="Permanent link">¶</a></h3>
<h3 id="htmlxhtml">HTML/XHTML常见问题<a class="headerlink" href="#htmlxhtml" title="Permanent link">¶</a></h3>
<h3 id="_146">安全注意事项<a class="headerlink" href="#_146" title="Permanent link">¶</a></h3>
<h3 id="flaskunicode">Flask中的Unicode<a class="headerlink" href="#flaskunicode" title="Permanent link">¶</a></h3>
<h3 id="flask_8">Flask扩展开发<a class="headerlink" href="#flask_8" title="Permanent link">¶</a></h3>
<h3 id="pocoo">Pocoo风格指引<a class="headerlink" href="#pocoo" title="Permanent link">¶</a></h3>
<h3 id="python3">Python3支持<a class="headerlink" href="#python3" title="Permanent link">¶</a></h3>
</article>
</div>
</div>
</main>
<footer class="md-footer">
<div class="md-footer-nav">
<nav class="md-footer-nav__inner md-grid">
<a href="../python/" title="Python" class="md-flex md-footer-nav__link md-footer-nav__link--prev" rel="prev">
<div class="md-flex__cell md-flex__cell--shrink">
<i class="md-icon md-icon--arrow-back md-footer-nav__button"></i>
</div>
<div class="md-flex__cell md-flex__cell--stretch md-footer-nav__title">
<span class="md-flex__ellipsis">
<span class="md-footer-nav__direction">
Previous
</span>
Python
</span>
</div>
</a>
<a href="../pyqt/" title="PyQt/Qt" class="md-flex md-footer-nav__link md-footer-nav__link--next" rel="next">
<div class="md-flex__cell md-flex__cell--stretch md-footer-nav__title">
<span class="md-flex__ellipsis">
<span class="md-footer-nav__direction">
Next
</span>
PyQt/Qt
</span>
</div>
<div class="md-flex__cell md-flex__cell--shrink">
<i class="md-icon md-icon--arrow-forward md-footer-nav__button"></i>
</div>
</a>
</nav>
</div>
<div class="md-footer-meta md-typeset">
<div class="md-footer-meta__inner md-grid">
<div class="md-footer-copyright">
<div class="md-footer-copyright__highlight">
Copyright © 2017 Yingxuanxuan.com
</div>
powered by
<a href="http://www.mkdocs.org" title="MkDocs">MkDocs</a>
and
<a href="http://squidfunk.github.io/mkdocs-material/" title="Material for MkDocs">
Material for MkDocs</a>
</div>
<div class="md-footer-social">
<link rel="stylesheet" href="https://cdnjs.cloudflare.com/ajax/libs/font-awesome/4.7.0/css/font-awesome.min.css">
<a href="https://github.com/yingxuanxuan" class="md-footer-social__link fa fa-github"></a>
<a href="https://twitter.com/yingxuanxuan_g" class="md-footer-social__link fa fa-twitter"></a>
</div>
</div>
</div>
</footer>
</div>
<script src="../assets/javascripts/application-f7ac33b6fb.js"></script>
<script>app.initialize({url:{base:".."}})</script>
<script>!function(e,t,a,n,o,c,i){e.GoogleAnalyticsObject=o,e[o]=e[o]||function(){(e[o].q=e[o].q||[]).push(arguments)},e[o].l=1*new Date,c=t.createElement(a),i=t.getElementsByTagName(a)[0],c.async=1,c.src=n,i.parentNode.insertBefore(c,i)}(window,document,"script","https://www.google-analytics.com/analytics.js","ga"),ga("create","UA-92401539-1","auto"),ga("set","anonymizeIp",!0),ga("send","pageview");var links=document.getElementsByTagName("a");Array.prototype.map.call(links,function(e){e.host!=document.location.host&&e.addEventListener("click",function(){var t=e.getAttribute("data-md-action")||"follow";ga("send","event","outbound",t,e.href)})});var query=document.forms.search.query;query.addEventListener("blur",function(){if(this.value){var e=document.location.pathname;ga("send","pageview",e+"?q="+this.value)}})</script>
</body>
</html> |