來源：伯樂在線專欄作者 - wklken
鏈接：http://python.jobbole.com/84003/
flask 使用的一些整理
最簡單的hello world
#!/usr/bin/env python
# encoding: utf-8
from flask import Flask
app = Flask(__name__)
@app.route('/')
def index():
return 'hello world'
if __name__ == '__main__':
app.run(debug=True)
#app.run(host='127.0.0.1', port=8000)
之后，訪問http://localhost:5000
支持post/get提交
@app.route('/', methods=['GET', 'POST'])
多個url指向
@app.route('/')
@app.route('/index')
不管post/get使用統(tǒng)一的接收
from flask import request
args = request.args if request.method == 'GET' else request.form
a = args.get('a', 'default')
處理json請求
request的header中
'Content-Type': 'application/json'
處理時:
data = request.get_json(silent=False)
獲取post提交中的checkbox
{%for page in pages %}
tr>td>input type=checkbox name=do_delete value='{{ page['id'] }}'>td>td>
{%endfor%}
page_ids = request.form.getlist('do_delete')
使用url中的參數(shù)
@app.route('/query/
/' )def query(qid):
pass
在request開始結(jié)束dosomething
一般可以處理數(shù)據(jù)庫連接等等
from flask import g
app = .....
@app.before_request
def before_request():
g.session = create_session()
@app.teardown_request
def teardown_request(exception):
g.session.close()
注冊Jinja2模板中使用的過濾器
@app.template_filter('reverse')
def reverse_filter(s):
return s[::-1]
或者
def reverse_filter(s):
return s[::-1]
app.jinja_env.filters['reverse'] = reverse_filter
可以這么用
def a():...
def b():...
FIL = {'a': a, 'b':b}
app.jinja_env.filters.update(FIL)
注冊Jinja2模板中使用的全局變量
JINJA2_GLOBALS = {'MEDIA_PREFIX': '/media/'}
app.jinja_env.globals.update(JINJA2_GLOBALS)
定義應(yīng)用使用的template和static目錄
app = Flask(__name__, template_folder=settings.TEMPLATE_FOLDER, static_folder = settings.STATIC_PATH)
使用Blueprint
from flask import Blueprint
bp_test = Blueprint('test', __name__)
#bp_test = Blueprint('test', __name__, url_prefix='/abc')
@bp_test.route('/')
--------
from xxx import bp_test
app = Flask(__name__)
app.register_blueprint(bp_test)
實例:
bp_video = Blueprint('video', __name__, url_prefix='/kw_news/video')
@bp_video.route('/search/category/', methods=['POST', 'GET'])
#注意這種情況下Blueprint中url_prefix不能以 '/' 結(jié)尾, 否則404
使用session
包裝cookie實現(xiàn)的，沒有session id
app.secret_key = 'PS#yio`%_!((f_or(%)))s'
# 然后
from flask import session
session['somekey'] = 1
session.pop('logged_in', None)
session.clear()
#過期時間,通過cookie實現(xiàn)的
from datetime import timedelta
session.permanent = True
app.permanent_session_lifetime = timedelta(minutes=5)
反向路由
from flask import url_for, render_template
@app.route('/')
def home():
login_uri = url_for('login', next=url_for('home'))
return render_template('home.html', **locals())
上傳文件
form action='/image/upload/' method='post' enctype='multipart/form-data'>
input type='file' name='upload' />
接收
f = request.files.get('upload')
img_data = f.read()
直接返回某個文件
return send_file(settings.TEMPLATE_FOLDER + 'tweet/tweet_list.html')
請求重定向
文檔（http://flask.pocoo.org/docs/api/#flask.redirect）
flask.redirect(location, code=302) the redirect status code. defaults to 302.Supported codes are 301, 302, 303, 305, and 307. 300 is not supported.
@app.route('/')
def hello():
return redirect(url_for('foo'))
@app.route('/foo')
def foo():
return'Hello Foo!'
獲取用戶真實ip
從request.headers獲取
real_ip = request.headers.get('X-Real-Ip', request.remote_addr)
或者, 使用werkzeug的middleware 文檔
from werkzeug.contrib.fixers import ProxyFix
app.wsgi_app = ProxyFix(app.wsgi_app)
return json & jsonp
import json
from flask import jsonify, Response, json
data = [] # or others
return jsonify(ok=True, data=data)
jsonp_callback = request.args.get('callback', '')
if jsonp_callback:
return Response(
'%s(%s);' % (jsonp_callback, json.dumps({'ok': True, 'data':data})),
mimetype='text/javascript'
)
return ok_jsonify(data)
配置讀取方法
# create our little application :)
app = Flask(__name__)
# Load default config and override config from an environment variable
app.config.update(dict(
DATABASE='/tmp/flaskr.db',
DEBUG=True,
SECRET_KEY='development key',
USERNAME='admin',
PASSWORD='default'
))
app.config.from_envvar('FLASKR_SETTINGS', silent=True)
------------------
# configuration
DATABASE = '/tmp/minitwit.db'
PER_PAGE = 30
DEBUG = True
SECRET_KEY = 'development key'
# create our little application :)
app = Flask(__name__)
app.config.from_object(__name__)
app.config.from_envvar('MINITWIT_SETTINGS', silent=True)
幾個不常用的方法
from flask import abort, flash
abort
if not session.get('logged_in'):
abort(401)
flash
flash('New entry was successfully posted')
異步調(diào)用
想在flask的一個請求中處理異步, 除了使用消息系統(tǒng), 可以用簡單的線程處理
from threading import Thread
def async(f):
def wrapper(*args, **kwargs):
thr = Thread(target=f, args=args, kwargs=kwargs)
thr.start()
return wrapper
@async
def dosomething(call_args):
print call_args
in a request handler, call `dosomething`
error handler
@app.errorhandler(404)
def not_found_error(error):
return render_template('404.html'), 404
@app.errorhandler(500)
def internal_error(error):
db.session.rollback()
return render_template('500.html'), 500
項目配置
1.直接
app.config['HOST']='xxx.a.com'
print app.config.get('HOST')
2.環(huán)境變量
export MyAppConfig=/path/to/settings.cfg
app.config.from_envvar('MyAppConfig')
3.對象
class Config(object):
DEBUG = False
TESTING = False
DATABASE_URI = 'sqlite://:memory:'
class ProductionConfig(Config):
DATABASE_URI = 'mysql://user@localhost/foo'
app.config.from_object(ProductionConfig)
print app.config.get('DATABASE_URI') # mysql://user@localhost/foo
4.文件
# default_config.py
HOST = 'localhost'
PORT = 5000
DEBUG = True
app.config.from_pyfile('default_config.py')
EG. 一個create_app方法
from flask import Flask, g
def create_app(debug=settings.DEBUG):
app = Flask(__name__,
template_folder=settings.TEMPLATE_FOLDER,
static_folder=settings.STATIC_FOLDER)
app.register_blueprint(bp_test)
app.jinja_env.globals.update(JINJA2_GLOBALS)
app.jinja_env.filters.update(JINJA2_FILTERS)
app.secret_key = 'PO+_)(*&678OUIJKKO#%_!(((%)))'
@app.before_request
def before_request():
g.xxx = ... #do some thing
@app.teardown_request
def teardown_request(exception):
g.xxx = ... #do some thing
return app
app = create_app(settings.DEBUG)
host=settings.SERVER_IP
port=settings.SERVER_PORT
app.run(host=host, port=port)
