Github上最受欢迎的Python轻量级框架Flask入门
運行python app.py，打開瀏覽器訪問http://localhost:5000/就可以看到頁面輸出了Hello World!
flask的誕生于2010年的愚人節，本來它只是作者無意間寫的一個小玩具，沒想到它卻悄悄流行起來了。漫長的8年時間，flask一直沒有發布一個嚴肅的正式版本，但是卻不能阻擋它成了github上最受好評的Python Web框架。
flask內核內置了兩個最重要的組件，所有其它的組件都是通過易擴展的插件系統集成進來的。這兩個內置的組件分別是werkzeug和jinja2。
werkzeug是一個用于編寫Python WSGI程序的工具包，它的結構設計和代碼質量在開源社區廣受褒揚，其源碼被尊為Python技術領域最值得閱讀的開源庫之一。
# wsgi.py from werkzeug.wrappers import Request, Response def application(request):return Response('Hello World!')if __name__ == '__main__':from werkzeug.serving import run_simplerun_simple('localhost', 4000, application) 復制代碼運行python wsgi.py打開瀏覽器訪問http://localhost:4000/就可以看到頁面輸出了Hello World!
Have you looked at werkzeug.routing? It's hard to find anything that's simpler, more self-contained, or purer-WSGI than Werkzeug, in general — I'm 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.py import mathfrom flask import Flask, requestapp = Flask(__name__)@app.route("/pi") def pi():# 默認參數n = int(request.args.get('n', '100'))s = 0.0for i in range(1, n):s += 1.0/i/ireturn 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 math import threadingfrom flask import Flask, request from 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] = pidef get(self, n):with self.lock:return self.pis.get(n)cache = PiCache() def pi():n = int(request.args.get('n', '100'))result = cache.get(n)if result:return jsonify({"cached": True, "result": result})s = 0.0for i in range(1, n):s += 1.0/i/iresult = 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 math import redisfrom flask import Flask, request from flask.json import jsonifyapp = Flask(__name__)class PiCache(object):def __init__(self, client):self.client = clientdef 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:returnreturn float(result)client = redis.StrictRedis() cache = PiCache(client) def pi():n = int(request.args.get('n', '100'))result = cache.get(n)if result:return jsonify({"cached": True, "result": result})s = 0.0for i in range(1, n):s += 1.0/i/iresult = 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 math import redisfrom flask import Flask, request from flask.json import jsonify from flask.views import MethodViewapp = Flask(__name__)class PiCache(object):def __init__(self, client):self.client = clientdef 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:returnreturn float(result)client = redis.StrictRedis() cache = PiCache(client)class PiAPI(MethodView):def __init__(self, cache):self.cache = cachedef get(self, n):result = self.cache.get(n)if result:return jsonify({"cached": True, "result": result})s = 0.0for i in range(1, n):s += 1.0/i/iresult = math.sqrt(6*s)self.cache.set(n, result)return jsonify({"cached": False, "result": result})# as_view提供了參數可以直接注入到MethodView的構造器中 # 我們不再使用request.args，而是將參數直接放進URL里面，這就是RESTFUL風格的URL app.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 math import redisfrom flask import Flask from flask.json import jsonify from flask_classy import FlaskView, route # 擴展app = Flask(__name__)# pi的cache和fib的cache要分開 class PiCache(object):def __init__(self, client):self.client = clientdef 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:returnreturn 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:returnreturn float(result)client = redis.StrictRedis() cache = PiCache(client)class MathAPI(FlaskView):def pi(self, n):result = cache.get_pi(n)if result:return jsonify({"cached": True, "result": result})s = 0.0for i in range(1, n):s += 1.0/i/iresult = math.sqrt(6*s)cache.set_pi(n, result)return jsonify({"cached": False, "result": result})def fib(self, n):result, cached = self.get_fib(n)return jsonify({"cached": cached, "result": result})def get_fib(self, n): # 遞歸，n不能過大，否則會堆棧過深溢出stackoverflowif n == 0:return 0, Trueif n == 1:return 1, Trueresult = cache.get_fib(n)if result:return result, Trueresult = 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知乎專欄
總結
以上是生活随笔為你收集整理的Github上最受欢迎的Python轻量级框架Flask入门的全部內容，希望文章能夠幫你解決所遇到的問題。
- 上一篇： KubeSphere 3.3.0 离线安
- 下一篇： 统计学习方法+Python机器学习实践指