最簡單的hello world
|
1
2
3
4
5
6
7
8
9
10
11
12
13
|
#!/usr/bin/env python# encoding: utf-8app = 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提交
|
1
|
@app.route('/', methods=['GET', 'POST']) |
多個url指向
|
1
2
|
@app.route('/')@app.route('/index') |
不管post/get使用統(tǒng)一的接收
|
1
2
3
|
from flask import requestargs = request.args if request.method == 'GET' else request.forma = args.get('a', 'default') |
處理json請求
request的header中
|
1
|
"Content-Type": "application/json" |
處理時:
|
1
|
data = request.get_json(silent=False) |
獲取post提交中的checkbox
|
1
2
3
4
5
|
{%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ù)
|
1
2
3
|
@app.route('/query/<qid>/')def query(qid): pass |
在request開始結(jié)束dosomething
一般可以處理數(shù)據(jù)庫連接等等
|
1
2
3
4
5
6
7
8
9
10
11
|
from flask import gapp = .....@app.before_requestdef before_request(): g.session = create_session()@app.teardown_requestdef teardown_request(exception): g.session.close() |
注冊Jinja2模板中使用的過濾器
|
1
2
3
|
@app.template_filter('reverse')def reverse_filter(s): return s[::-1] |
或者
|
1
2
3
|
def reverse_filter(s): return s[::-1]app.jinja_env.filters['reverse'] = reverse_filter |
可以這么用
|
1
2
3
4
5
|
def a():...def b():...FIL = {'a': a, 'b':b}app.jinja_env.filters.update(FIL) |
注冊Jinja2模板中使用的全局變量
|
1
2
|
JINJA2_GLOBALS = {'MEDIA_PREFIX': '/media/'}app.jinja_env.globals.update(JINJA2_GLOBALS) |
定義應用使用的template和static目錄
|
1
|
app = Flask(__name__, template_folder=settings.TEMPLATE_FOLDER, static_folder = settings.STATIC_PATH) |
使用Blueprint
|
1
2
3
4
5
6
7
8
9
10
11
|
from flask import Blueprintbp_test = Blueprint('test', __name__)#bp_test = Blueprint('test', __name__, url_prefix='/abc')@bp_test.route('/')--------from xxx import bp_testapp = Flask(__name__)app.register_blueprint(bp_test) |
實例:
|
1
2
3
|
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
|
1
|
app.secret_key = 'PS#yio`%_!((f_or(%)))s' |
然后
|
1
2
3
4
5
6
7
8
9
10
11
|
from flask import sessionsession['somekey'] = 1session.pop('logged_in', None)session.clear()#過期時間,通過cookie實現(xiàn)的from datetime import timedeltasession.permanent = Trueapp.permanent_session_lifetime = timedelta(minutes=5) |
反向路由
|
1
2
3
4
5
6
|
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()) |
上傳文件
|
1
2
|
<form action="/image/upload/" method="post" enctype="multipart/form-data"><input type="file" name="upload" /> |
接收
|
1
2
|
f = request.files.get('upload')img_data = f.read() |
直接返回某個文件
|
1
|
return send_file(settings.TEMPLATE_FOLDER + 'tweet/tweet_list.html') |
請求重定向
|
1
2
3
4
5
6
7
8
9
|
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 文檔
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
|
from werkzeug.contrib.fixers import ProxyFixapp.wsgi_app = ProxyFix(app.wsgi_app)return json & jsonpimport jsonfrom flask import jsonify, Response, jsondata = [] # or othersreturn 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) |
配置讀取方法
|
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
|
# create our little application :)app = Flask(__name__)# Load default config and override config from an environment variableapp.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)------------------# configurationDATABASE = '/tmp/minitwit.db'PER_PAGE = 30DEBUG = TrueSECRET_KEY = 'development key'# create our little application :)app = Flask(__name__)app.config.from_object(__name__)app.config.from_envvar('MINITWIT_SETTINGS', silent=True) |
幾個不常用的方法
|
1
2
3
4
5
6
7
8
|
from flask import abort, flashabortif not session.get('logged_in'): abort(401)flashflash('New entry was successfully posted') |
異步調(diào)用
想在flask的一個請求中處理異步, 除了使用消息系統(tǒng), 可以用簡單的線程處理
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
|
from threading import Threaddef async(f): def wrapper(*args, **kwargs): thr = Thread(target=f, args=args, kwargs=kwargs) thr.start() return wrapper@asyncdef dosomething(call_args): print call_argsin 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.直接
|
1
2
|
app.config['HOST']='xxx.a.com'print app.config.get('HOST') |
2.環(huán)境變量
|
1
2
|
export MyAppConfig=/path/to/settings.cfgapp.config.from_envvar('MyAppConfig') |
3.對象
|
1
2
3
4
5
6
7
8
9
10
|
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.文件
|
1
2
3
4
5
6
|
# default_config.pyHOST = 'localhost'PORT = 5000DEBUG = Trueapp.config.from_pyfile('default_config.py') |
EG. 一個create_app方法
|
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
|
from flask import Flask, gdef 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 appapp = create_app(settings.DEBUG)host=settings.SERVER_IPport=settings.SERVER_PORTapp.run(host=host, port=port)change log:2013-09-09 create2014-10-25 update |
