Github上最受歡迎的Python輕量級框架Flask入門

來自專欄碼洞

flask最近終於發布了它的1.0版本更新，從項目開源到最近的1.0版本flask已經走過了8個年頭。

# app.pyfrom flask import Flaskapp = Flask(__name__)@app.route("/")def hello(): return "Hello World!"if __name__ == "__main__": app.run()

運行python app.py，打開瀏覽器訪問http://localhost:5000/就可以看到頁面輸出了Hello World!

flask的誕生於2010年的愚人節，本來它只是作者無意間寫的一個小玩具，沒想到它卻悄悄流行起來了。漫長的8年時間，flask一直沒有發布一個嚴肅的正式版本，但是卻不能阻擋它成了github上最受好評的Python Web框架。

flask內核內置了兩個最重要的組件，所有其它的組件都是通過易擴展的插件系統集成進來的。這兩個內置的組件分別是werkzeug和jinja2。

werkzeug是一個用於編寫Python WSGI程序的工具包，它的結構設計和代碼質量在開源社區廣受褒揚，其源碼被尊為Python技術領域最值得閱讀的開源庫之一。

# wsgi.pyfrom werkzeug.wrappers import Request, Response@Request.applicationdef application(request): return Response(Hello World!)if __name__ == __main__: from werkzeug.serving import run_simple run_simple(localhost, 4000, application)

運行python wsgi.py打開瀏覽器訪問http://localhost:4000/就可以看到頁面輸出了Hello World!

Have you looked at werkzeug.routing? Its hard to find anything thats simpler, more self-contained, or purer-WSGI than Werkzeug, in general — Im quite a fan of it!

by Alex Martelli, the author of 《Python in a Nutshell》 && 《Python Cookbook》

jinja2是一個功能極為強大的模板系統，它完美支持unicode中文，每個模板都運行在安全的沙箱環境中，使用jinja2編寫的模板代碼非常優美。

{% extends "layout.html" %}{% block body %} <ul> {% for user in users %} <li><a href="{{ user.url }}">{{ user.username }}</a></li> {% endfor %} </ul>{% endblock %}

werkzeug和jinja2這兩個庫的共同特點是編寫的代碼賞心悅目，作者Armin Ronacher選擇這兩個庫來作為flask的基石說明作者有非常挑剔的代碼品味。那麼作者是誰呢，鐺！他是一位來自澳大利亞的帥哥！

好，閑話少說言歸正傳，接下來我們開始體驗flask的神奇魅力。

安裝flask

pip install flask

圓周率計算API

圓周率可以使用正整數的平方倒數之和求得，當這個級數趨於無限時，值會越來越接近圓周率。

# flask_pi.pyimport mathfrom flask import Flask, requestapp = Flask(__name__)@app.route("/pi")def pi(): # 默認參數 n = int(request.args.get(n, 100)) s = 0.0 for i in range(1, n): s += 1.0/i/i return str(math.sqrt(6*s))if __name__ == __main__: app.run()

運行python flask_pi.py，打開瀏覽器訪問http://localhost:5000/pi?n=1000000，可以看到頁面輸出3.14159169866，這個值同圓周率已經非常接近。

注意pi()的返回值不能是浮點數，所以必須使用str轉換成字元串

再仔細觀察代碼，你還會注意到一個特殊的變數request，它看起來似乎是一個全局變數。從全局變數里拿當前請求參數，這非常奇怪。如果在多線程環境中，該如何保證每個線程拿到的都是當前線程正在處理的請求參數呢？所以它不能是全局變數，它是線程局部變數，線程局部變數外表上和全局變數沒有差別，但是在訪問線程局部變數時，每個線程得到的都是當前線程內部共享的對象。

緩存計算結果

為了避免重複計算，我們將已經計算的pi(n)值緩存起來，下次就可以直接查詢。同時我們不再只返回一個單純的字元串，我們返回一個json串，裡面有一個欄位cached用來標識當前的結果是否從緩存中直接獲取的。

import mathimport threadingfrom flask import Flask, requestfrom flask.json import jsonifyapp = Flask(__name__)class PiCache(object): def __init__(self): self.pis = {} self.lock = threading.RLock() def set(self, n, pi): with self.lock: self.pis[n] = pi def get(self, n): with self.lock: return self.pis.get(n)cache = PiCache()@app.route("/pi")def pi(): n = int(request.args.get(n, 100)) result = cache.get(n) if result: return jsonify({"cached": True, "result": result}) s = 0.0 for i in range(1, n): s += 1.0/i/i result = math.sqrt(6*s) cache.set(n, result) return jsonify({"cached": False, "result": result})if __name__ == __main__: app.run()

運行python flask_pi.py，打開瀏覽器訪問http://localhost:5000/pi?n=1000000，可以看到頁面輸出

{ "cached": false, "result": 3.141591698659554}

再次刷新頁面，我們可以觀察到cached欄位變成了true，說明結果確實已經緩存了

{ "cached": true, "result": 3.141591698659554}

讀者也許會問，為什麼緩存類PiCache需要使用RLock呢？這是因為考慮到多線程環境下Python的字典讀寫不是完全線程安全的，需要使用鎖來保護一下數據結構。

分散式緩存

上面的緩存僅僅是內存緩存，進程重啟後，緩存結果消失，下次計算又得重新開始。

if __name__ == __main__: app.run(127.0.0.1, 5001)

如果開啟第二個埠5001來提供服務，那這第二個進程也無法享受第一個進程的內存緩存，而必須重新計算。所以這裡要引入分散式緩存Redis來共享計算緩存，避免跨進程重複計算，避免重啟重新計算。

import mathimport redisfrom flask import Flask, requestfrom flask.json import jsonifyapp = Flask(__name__)class PiCache(object): def __init__(self, client): self.client = client def set(self, n, result): self.client.hset("pis", str(n), str(result)) def get(self, n): result = self.client.hget("pis", str(n)) if not result: return return float(result)client = redis.StrictRedis()cache = PiCache(client)@app.route("/pi")def pi(): n = int(request.args.get(n, 100)) result = cache.get(n) if result: return jsonify({"cached": True, "result": result}) s = 0.0 for i in range(1, n): s += 1.0/i/i result = math.sqrt(6*s) cache.set(n, result) return jsonify({"cached": False, "result": result})if __name__ == __main__: app.run(127.0.0.1, 5000)

運行python flask_pi.py，打開瀏覽器訪問http://localhost:5000/pi?n=1000000，可以看到頁面輸出

{ "cached": false, "result": 3.141591698659554}

再次刷新頁面，我們可以觀察到cached欄位變成了true，說明結果確實已經緩存了

{ "cached": true, "result": 3.141591698659554}

重啟進程，再次刷新頁面，可以看書頁面輸出的cached欄位依然是true，說明緩存結果不再因為進程重啟而丟失。

MethodView

寫過Django的朋友們可能會問，Flask是否支持類形式的API編寫方式，回答是肯定的。下面我們使用Flask原生支持的MethodView來改寫一下上面的服務。

import mathimport redisfrom flask import Flask, requestfrom flask.json import jsonifyfrom flask.views import MethodViewapp = Flask(__name__)class PiCache(object): def __init__(self, client): self.client = client def set(self, n, result): self.client.hset("pis", str(n), str(result)) def get(self, n): result = self.client.hget("pis", str(n)) if not result: return return float(result)client = redis.StrictRedis()cache = PiCache(client)class PiAPI(MethodView): def __init__(self, cache): self.cache = cache def get(self, n): result = self.cache.get(n) if result: return jsonify({"cached": True, "result": result}) s = 0.0 for i in range(1, n): s += 1.0/i/i result = math.sqrt(6*s) self.cache.set(n, result) return jsonify({"cached": False, "result": result})# as_view提供了參數可以直接注入到MethodView的構造器中# 我們不再使用request.args，而是將參數直接放進URL裡面，這就是RESTFUL風格的URLapp.add_url_rule(/pi/<int:n>, view_func=PiAPI.as_view(pi, cache))if __name__ == __main__: app.run(127.0.0.1, 5000)

我們實現了MethodView的get方法，說明該API僅支持HTTP請求的GET方法。如果要支持POST、PUT和DELETE方法，需要用戶自己再去實現這些方法。

flask默認的MethodView挺好用，但是也不夠好用，它無法在一個類里提供多個不同URL名稱的API服務。所以接下來我們引入flask的擴展flask-classy來解決這個問題。

小試flask擴展flask-classy

使用擴展的第一步是安裝擴展pip install flask-classy，然後我們在同一個類里再加一個新的API服務，計算斐波那契級數。

import mathimport redisfrom flask import Flaskfrom flask.json import jsonifyfrom flask_classy import FlaskView, route # 擴展app = Flask(__name__)# pi的cache和fib的cache要分開class PiCache(object): def __init__(self, client): self.client = client def set_fib(self, n, result): self.client.hset("fibs", str(n), str(result)) def get_fib(self, n): result = self.client.hget("fibs", str(n)) if not result: return return int(result) def set_pi(self, n, result): self.client.hset("pis", str(n), str(result)) def get_pi(self, n): result = self.client.hget("pis", str(n)) if not result: return return float(result)client = redis.StrictRedis()cache = PiCache(client)class MathAPI(FlaskView): @route("/pi/<int:n>") def pi(self, n): result = cache.get_pi(n) if result: return jsonify({"cached": True, "result": result}) s = 0.0 for i in range(1, n): s += 1.0/i/i result = math.sqrt(6*s) cache.set_pi(n, result) return jsonify({"cached": False, "result": result}) @route("/fib/<int:n>") def fib(self, n): result, cached = self.get_fib(n) return jsonify({"cached": cached, "result": result}) def get_fib(self, n): # 遞歸，n不能過大，否則會堆棧過深溢出stackoverflow if n == 0: return 0, True if n == 1: return 1, True result = cache.get_fib(n) if result: return result, True result = self.get_fib(n-1)[0] + self.get_fib(n-2)[0] cache.set_fib(n, result) return result, FalseMathAPI.register(app, route_base=/) # 註冊到appif __name__ == __main__: app.run(127.0.0.1, 5000)

訪問http://localhost:5000/fib/100，我們可以看到頁面輸出了

{ "cached": false, "result": 354224848179261915075}

訪問http://localhost:5000/pi/10000000，計算量比較大，所以多轉了一回，最終頁面輸出了

{ "cached": false, "result": 3.141592558095893}

閱讀更多高級文章，關注微信訂閱號「碼洞」

擴展閱讀

廖雪峰教你ThreadLocal的正確用法

Python字典是否是線程安全的

Hello Flask知乎專欄