從 Flask 切到 FastAPI 后,起飛了!
共 21148字,需瀏覽 43分鐘
·
2024-07-20 08:30
↓推薦關注↓
剛好筆者這幾天上手體驗 FastAPI,感受到這個框架易用和方便。之前也使用過 Python 中的 Django 和 Flask 作為項目的框架。Django 說實話上手也方便,但是學習起來有點重量級框架的感覺,F(xiàn)astAPI 帶給我的直觀體驗還是很輕便的,本文翻譯的這篇文章就會著重介紹 FastAPI 和 Flask 的區(qū)別。
Python 是最流行的編程語言之一。從腳本到 API 開發(fā)再到機器學習,Python 都有著它自己的足跡。因為 Python 注重開發(fā)者的體驗和其所能提供的大量工具而大受歡迎。網絡框架 Flask 就是這樣一個工具,它在機器學習社區(qū)中很受歡迎。它也被廣泛用于 API開發(fā)。但是有一個新的框架正在崛起: FastAPI。與 Flask 不同,F(xiàn)astAPI 是一個 ASGI(Asynchronous Server Gateway Interface 異步服務器網關接口)框架。與 Go 和 NodeJS 一樣,F(xiàn)astAPI 是最快的基于 Python 的 Web 框架之一。
本文針對那些有興趣從 Flask 轉移到 FastAPI 的人,比較和對比了 Flask 和 FastAPI 的常見模式。
# FastAPI vs Flask
FastAPI 的構建考慮了以下三個主要問題:
速度
開發(fā)者經驗
開放標準
你可以把 FastAPI 看作是把 Starlette、Pydantic、OpenAPI 和 JSON Schema 粘合在一起的膠水。
本質上說,F(xiàn)astAPI 使用 Pydantic 進行數(shù)據(jù)驗證,并使用 Starlette 作為工具,使其與 Flask 相比快得驚人,具有與 Node 或 Go 中的高速 Web APIs 相同的性能。
Starlette + Uvicorn 提供異步請求能力,這是 Flask 所缺乏的。
有了 Pydantic 以及類型提示,你就可以得到一個具有自動完成功能的良好的編輯體驗。你還可以得到數(shù)據(jù)驗證、序列化和反序列化(用于構建一個 API),以及自動化文檔(通過 JSON Schema 和 OpenAPI )。
也就是說,F(xiàn)lask 的使用更為廣泛,所以它經過了實戰(zhàn)檢驗,并且有更大的社區(qū)支持它。由于這兩個框架都是用來擴展的,F(xiàn)lask 顯然是贏家,因為它有龐大的插件生態(tài)系統(tǒng)。
建議:
如果你對上述三個問題有共鳴,厭倦了 Flask 擴展時的大量選擇,希望利用異步請求,或者只是想建立一個 RESTful API,請使用 FastAPI。
如果你對 FastAPI 的成熟度不滿意,需要用服務器端模板構建一個全棧應用,或者離不開一些社區(qū)維護的 Flask 擴展,就可以使用 Flask。
# 開始
安裝
與任何其他 Python 包一樣,安裝非常簡單。
Flask
pip install flask
# or
poetry add flask
pipenv install flask
conda install flask
FastAPI
pip install fastapi uvicorn
# or
poetry add fastapi uvicorn
pipenv install fastapi uvicorn
conda install fastapi uvicorn -c conda-forge
與 Flask 不同,F(xiàn)astAPI 沒有內置的開發(fā)服務器,因此需要像 Uvicorn 或 Daphne 這樣的 ASGI 服務器。
"Hello World" 應用
Flask
# flask_code.py
from flask import Flask
app = Flask(__name__)
@app.route("/")
def home():
return {"Hello": "World"}
if __name__ == "__main__":
app.run()
FastAPI
# fastapi_code.py
import uvicorn
from fastapi import FastAPI
app = FastAPI()
@app.get("/")
def home():
return {"Hello": "World"}
if __name__ == "__main__":
uvicorn.run("fastapi_code:app")
像 reload=True 這樣的參數(shù)可以被傳遞到 uvicorn.run() 中,以實現(xiàn)開發(fā)時的熱重載。
或者,您可以直接從終端啟動服務器:
uvicorn run fastapi_code:app
熱加載模式:
uvicorn run fastapi_code:app --reload
# 配置
Flask 和 FastAPI 都提供了許多選項來處理不同環(huán)境的不同配置。兩者都支持以下模式:
環(huán)境變量
配置文件
實例文件夾
類和繼承
有關更多信息,請參閱其各自的文檔:
Flask: https://flask.palletsprojects.com/en/2.0.x/config/
FastAPI:https://fastapi.tiangolo.com/advanced/settings/
Flask
import os
from flask import Flask
class Config(object):
MESSAGE = os.environ.get("MESSAGE")
app = Flask(__name__)
app.config.from_object(Config)
@app.route("/settings")
def get_settings():
return { "message": app.config["MESSAGE"] }
if __name__ == "__main__":
app.run()
現(xiàn)在,在你運行服務器之前,設置適當?shù)沫h(huán)境變量:
export MESSAGE="hello, world"
FastAPI
import uvicorn
from fastapi import FastAPI
from pydantic import BaseSettings
class Settings(BaseSettings):
message: str
settings = Settings()
app = FastAPI()
@app.get("/settings")
def get_settings():
return { "message": settings.message }
if __name__ == "__main__":
uvicorn.run("fastapi_code:app")
同樣,在運行服務器之前,設置適當?shù)沫h(huán)境變量:
export MESSAGE="hello, world"
# 路由, 模板和視圖
HTTP 方法
Flask
from flask import request
@app.route("/", methods=["GET", "POST"])
def home():
# handle POST
if request.method == "POST":
return {"Hello": "POST"}
# handle GET
return {"Hello": "GET"}
FastAPI
@app.get("/")
def home():
return {"Hello": "GET"}
@app.post("/")
def home_post():
return {"Hello": "POST"}
FastAPI 為每個方法提供單獨的裝飾器:
@app.get("/")
@app.post("/")
@app.delete("/")
@app.patch("/")
URL 參數(shù)
通過 URL(如 /employee/1 )傳遞信息以管理狀態(tài):
Flask
@app.route("/employee/<int:id>")
def home():
return {"id": id}
FastAPI
@app.get("/employee/{id}")
def home(id: int):
return {"id": id}
URL參數(shù)的指定類似于一個 f-string 表達式。此外,你還可以利用類型提示。這里,我們在運行時告訴 Pydantic, id 是 int 類型的。在開發(fā)中,這也可以幫助完成更好的代碼完成度。
查詢參數(shù)
與 URL 參數(shù)一樣,查詢參數(shù)(如 /employee?department=sales )也可用于管理狀態(tài)(通常用于過濾或排序):
Flask
from flask import request
@app.route("/employee")
def home():
department = request.args.get("department")
return {"department": department}
FastAPI
@app.get("/employee")
def home(department: str):
return {"department": department}
模板
Flask
from flask import render_template
@app.route("/")
def home():
return render_template("index.html")
默認情況下,F(xiàn)lask會在 "templates "文件夾中尋找模板。
FastAPI
你需要安裝 Jinja:
pip install jinja2
實現(xiàn):
from fastapi import Request
from fastapi.templating import Jinja2Templates
from fastapi.responses import HTMLResponse
app = FastAPI()
templates = Jinja2Templates(directory="templates")
@app.get("/", response_class=HTMLResponse)
def home(request: Request):
return templates.TemplateResponse("index.html", {"request": request})
對于 FastAPI,你需要明確地定義 "模板 "文件夾。然后對于每個響應,需要提供請求上下文。
靜態(tài)文件
Flask
默認情況下,F(xiàn)lask 從“static”文件夾中提供靜態(tài)文件。
FastAPI
在 FastAPI 中,需要為靜態(tài)文件掛載一個文件夾:
from fastapi.staticfiles import StaticFiles
app = FastAPI()
app.mount("/static", StaticFiles(directory="static"), name="static")
異步任務
Flask
從 Flask 2.0 開始,您可以使用 async/await 創(chuàng)建異步路由處理程序:
@app.route("/")
async def home():
result = await some_async_task()
return result
有關 Flask 中異步視圖的更多信息,請查看 Flask 2.0 中的異步一文。
Flask 中的異步也可以通過使用線程(并發(fā))或多處理(并行)或 Celery 或 RQ 等工具來實現(xiàn):
Asynchronous Tasks with Flask and Celery:https://testdriven.io/blog/flask-and-celery/
Asynchronous Tasks with Flask and Redis Queue:https://testdriven.io/blog/asynchronous-tasks-with-flask-and-redis-queue/
FastAPI
由于 FastAPI 對 asyncio 的原生支持,它極大地簡化了異步任務。要使用的話,只需在視圖函數(shù)中添加 async 關鍵字:
@app.get("/")
async def home():
result = await some_async_task()
return result
FastAPI 還具有后臺任務功能,您可以使用它來定義返回響應后要運行的后臺任務。這對于不需要在發(fā)送回響應之前完成的操作很有用。
from fastapi import BackgroundTasks
def process_file(filename: str):
# process file :: takes minimum 3 secs (just an example)
pass
@app.post("/upload/{filename}")
async def upload_and_process(filename: str, background_tasks: BackgroundTasks):
background_tasks.add_task(process_file, filename)
return {"message": "processing file"}
在這里,響應將被即時發(fā)送,而不會讓用戶等待文件處理完成。
當你需要進行繁重的后臺計算時,或者你需要一個任務隊列來管理任務(tasks)和工作者(workers)時,你可能想使用Celery 而不是 BackgroundTasks。更多內容請參考 FastAPI 和 Celery 的異步任務:https://testdriven.io/blog/fastapi-and-celery/
依賴注入
Flask
雖然你可以實現(xiàn)自己的依賴注入解決方案,但 Flask 默認沒有真正的一流支持。相反,你需要使用一個外部包,如 flask-injector。
FastAPI
另一方面,F(xiàn)astAPI 具有處理依賴注入的強大解決方案。
例如:
from databases import Database
from fastapi import Depends
from starlette.requests import Request
from db_helpers import get_all_data
def get_db(request: Request):
return request.app.state._db
@app.get("/data")
def get_data(db: Database = Depends(get_db)):
return get_all_data(db)
因此,get_db 將獲取對在應用程序的啟動事件處理程序中創(chuàng)建的數(shù)據(jù)庫連接的引用。 Depends 然后用于向 FastAPI 指示路由“依賴于” get_db。因此,它應該在路由處理程序中的代碼之前執(zhí)行,并且結果應該“注入”到路由本身。
數(shù)據(jù)校驗
Flask
Flask 沒有任何內部數(shù)據(jù)驗證支持。您可以使用功能強大的 Pydantic 包通過 Flask-Pydantic 進行數(shù)據(jù)驗證。
FastAPI
FastAPI 如此強大的原因之一是它支持 Pydantic。
from pydantic import BaseModel
app = FastAPI()
class Request(BaseModel):
username: str
password: str
@app.post("/login")
async def login(req: Request):
if req.username == "testdriven.io" and req.password == "testdriven.io":
return {"message": "success"}
return {"message": "Authentication Failed"}
在這里,我們接受一個模型 Request 的輸入。該 payload 必須包含一個用戶名和密碼。
# correct payload format
? curl -X POST 'localhost:8000/login' \
--header 'Content-Type: application/json' \
--data-raw '{"username": "testdriven.io","password":"testdriven.io"}'
{"message":"success"}
# incorrect payload format
? curl -X POST 'localhost:8000/login' \
--header 'Content-Type: application/json' \
--data-raw '{"username": "testdriven.io","passwords":"testdriven.io"}'
{"detail":[{"loc":["body","password"],"msg":"field required","type":"value_error.missing"}]}
注意到這個請求。我們把密碼 passwords 作為一個鍵而不是 password 傳遞進去。Pydantic 模型會自動告訴用戶,password 字段是缺失的。
序列化和反序列化
Flask
最簡單的序列化方法是使用 jsonify:
from flask import jsonify
from data import get_data_as_dict
@app.route("/")
def send_data():
return jsonify(get_data_as_dict)
對于復雜的對象,F(xiàn)lask 開發(fā)者經常使用 Flask-Marshmallow
FastAPI
FastAPI 自動序列化任何返回的字典 dict 。對于更復雜和結構化的數(shù)據(jù),使用 Pydantic:
from pydantic import BaseModel
app = FastAPI()
class Request(BaseModel):
username: str
email: str
password: str
class Response(BaseModel):
username: str
email: str
@app.post("/login", response_model=Response)
async def login(req: Request):
if req.username == "testdriven.io" and req.password == "testdriven.io":
return req
return {"message": "Authentication Failed"}
在這里,我們添加了一個包含三個輸入的 Request 模型:用戶名、電子郵件和密碼。我們還定義了一個僅包含用戶名和電子郵件的 Response 模型。輸入 Request 模型處理反序列化,而輸出 Response 模型處理對象序列化。然后通過 response_model 參數(shù)將響應模型傳遞給裝飾器。
現(xiàn)在,如果我們將請求本身作為響應返回,Pydantic 將省略 password ,因為我們定義的響應模型不包含密碼字段。
例如:
# output
? curl -X POST 'localhost:8000/login' \
--header 'Content-Type: application/json' \
--data-raw '{"username":"testdriven.io","email":"[email protected]","password":"testdriven.io"}'
{"username":"testdriven.io","email":"[email protected]"}
中間件
中間件被用來在每個請求被視圖功能處理之前應用邏輯。
Flask
class middleware:
def __init__(self, app) -> None:
self.app = app
def __call__(self, environ, start_response):
start = time.time()
response = self.app(environ, start_response)
end = time.time() - start
print(f"request processed in {end} s")
return response
app = Flask(__name__)
app.wsgi_app = middleware(app.wsgi_app)
FastAPI
from fastapi import Request
@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
start_time = time.time()
response = await call_next(request)
process_time = time.time() - start_time
print(f"request processed in {process_time} s")
return response
@app.middleware("http") 裝飾器是在 FastAPI 中創(chuàng)建中間件的必備工具。上述中間件計算處理請求所花費的時間。視圖函數(shù)處理請求后,計算總處理時間并將其作為響應頭返回。
# flask output(logs)
request processed in 0.0010077953338623047 s
127.0.0.1 - - [22/Sep/2020 18:56:21] "GET / HTTP/1.1" 200 -
# fastapi output(logs)
request processed in 0.0009925365447998047 s
INFO: 127.0.0.1:51123 - "GET / HTTP/1.1" 200 OK
模塊化
隨著應用程序的發(fā)展,在某些時候你會想把類似的視圖、模板、靜態(tài)文件和模型組合在一起,以幫助把應用程序分解成更小的組件。
Flask
在 Flask 中,藍圖被用來實現(xiàn)模塊化:
# blueprints/product/views.py
from flask import Blueprint
product = Blueprint("product", __name__)
@product.route("/product1")
...
# main.py
from blueprints.product.views import product
app.register_blueprint(product)
FastAPI
同時,在 FastAPI 中,模塊化是通過 APIRouter 實現(xiàn)的:
# routers/product/views.py
from fastapi import APIRouter
product = APIRouter()
@product.get("/product1")
...
# main.py
from routers.product.views import product
app.include_router(product)
# 其他特點
自動文檔
Flask
Flask 不會自動創(chuàng)建開箱即用的 API 文檔。然而,有幾個擴展可以處理這個問題,比如 flask-swagger 和 Flask RESTX,但它們需要額外的設置。
FastAPI
默認情況下,F(xiàn)astAPI 支持 OpenAPI 以及 Swagger UI 和 ReDoc。這意味著每個端點都自動從與端點關聯(lián)的元數(shù)據(jù)中記錄下來。
此處列出了所有已注冊的端點
替代文檔
管理應用
Flask
Flask 有一個廣泛使用的第三方管理包,稱為 Flask-Admin,用于快速對您的模型執(zhí)行 CRUD 操作。
FastAPI
截至目前,有兩個流行的 FastAPI 擴展用于此:
FastAPI Admin - 功能性管理面板,提供用于對數(shù)據(jù)執(zhí)行 CRUD 操作的用戶界面。
SQLAlchemy Admin -FastAPI/Starlette 的管理面板,可與 SQLAlchemy 模型一起使用。
身份認證
Flask
雖然 Flask 沒有原生解決方案,但可以使用多個第三方擴展。
FastAPI
FastAPI 通過 fastapi.security 包原生支持許多安全和身份驗證工具。通過幾行代碼,您可以將基本的 HTTP 身份驗證添加到您的應用程序中:
import secrets
from fastapi import Depends, FastAPI, HTTPException, status
from fastapi.security import HTTPBasic, HTTPBasicCredentials
app = FastAPI()
security = HTTPBasic()
def get_current_username(credentials: HTTPBasicCredentials = Depends(security)):
correct_username = secrets.compare_digest(credentials.username, "stanleyjobson")
correct_password = secrets.compare_digest(credentials.password, "swordfish")
if not (correct_username and correct_password):
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED)
return credentials.username
@app.get("/whoami")
def who_ami_i(username: str = Depends(get_current_username)):
return {"username": username}
FastAPI 通過 OpenAPI 標準實現(xiàn) OAuth2 和 OpenID Connect。
查看官方文檔中的以下資源以獲取更多信息:
Security Intro:https://fastapi.tiangolo.com/tutorial/security/
Advanced Security:https://fastapi.tiangolo.com/advanced/security/
其他資源
Web Authentication Methods Compared:https://testdriven.io/blog/web-authentication-methods/
Adding Social Authentication to Flask:https://testdriven.io/blog/flask-social-auth/
Session-based Auth with Flask for Single Page Apps:https://testdriven.io/blog/flask-spa-auth/
Securing FastAPI with JWT Token-based Authentication:https://testdriven.io/blog/fastapi-jwt-auth/
CORS
CORS(跨源資源共享)中間件檢查請求是否來自允許的來源。如果是,則將請求傳遞給下一個中間件或視圖函數(shù)。如果不是,它會拒絕請求,并將錯誤響應發(fā)送回調用者。
Flask Flask 需要一個名為 Flask-CORS 的外部包來支持 CORS:
pip install flask-cors
基本實現(xiàn):
from flask_cors import CORS
app = Flask(__name__)
CORS(app)
FastAPI
FastAPI 原生支持 CORS:
from fastapi.middleware.cors import CORSMiddleware
app = FastAPI()
origins = ["*"]
app.add_middleware(CORSMiddleware, allow_origins=origins)
# 測試
Flask
import pytest
from flask import Flask
app = Flask(__name__)
@app.route("/")
def home():
return {"message": "OK"}
def test_hello():
res = app.test_client().get("/")
assert res.status_code == 200
assert res.data == b'{"message":"OK"}\n'
FastAPI
from fastapi import FastAPI
from fastapi.testclient import TestClient
app = FastAPI()
@app.get("/")
async def home():
return {"message": "OK"}
client = TestClient(app)
def test_home():
res = client.get("/")
assert res.status_code == 200
assert res.json() == {"message": "OK"}
FastAPI 提供了一個 TestClient。有了它,你可以直接用 FastAPI 運行 pytest。有關更多信息,請查看官方文檔中的測試指南。
# 部署
生產服務器
Flask
Flask 默認運行開發(fā) WSGI(Web 服務器網關接口)應用程序服務器。對于生產環(huán)境,您需要使用生產級 WSGI 應用服務器,例如 Gunicorn、uWSGI 或 mod_wsgi
安裝 Gunicorn:
pip install gunicorn
啟動服務:
# main.py
# app = Flask(__name__)
gunicorn main:app
FastAPI
由于 FastAPI 沒有開發(fā)服務器,您將使用 Uvicorn(或 Daphne)進行開發(fā)和生產。
安裝 Uvicorn:
pip install uvicorn
啟動服務:
python
# main.py
# app = FastAPI()
uvicorn main:app
您可能希望使用 Gunicorn 來管理 Uvicorn,以便同時利用并發(fā)性(通過 Uvicorn)和并行性(通過 Gunicorn worker):
# main.py
# app = FastAPI()
gunicorn -w 3 -k uvicorn.workers.UvicornWorker main:app
Docker
Flask
FROM python3.10-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install -r requirements.txt
COPY . .
EXPOSE 5000
CMD ["gunicorn", "main:app"]
這是 Flask 最簡單的 Dockerfile 之一。要了解如何針對生產對其進行全面配置,請查看使用 Postgres、Gunicorn 和 Nginx 教程對 Flask 進行 Docker 化。
FastAPI
sql
FROM python3.10-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install -r requirements.txt
COPY . .
EXPOSE 8000
CMD ["uvicorn", "main:app"]
同樣,這是一個非常簡單的配置。 FastAPI 作者提供了幾個生產就緒的 Dockerfile。有關更多信息,請查看官方 FastAPI 文檔以及 Dockerizing FastAPI with Postgres、Uvicorn 和 Traefik 教程。
# 總結
退一步講,Django 和 Flask 是兩個最流行的基于 Python 的網絡框架(FastAPI 是第三大流行框架)。不過它們(Django 和 Flask)的理念非常不同。Flask 比 Django 的優(yōu)勢在于 Flask 是一個微框架。程序結構由程序員自己決定,不強制執(zhí)行。開發(fā)者可以在他們認為合適的時候添加第三方擴展來改進他們的代碼。也就是說,通常情況下,隨著代碼庫的增長,需要一些幾乎所有網絡應用都需要的通用功能。這些功能與框架的緊密結合,使得終端開發(fā)者需要自己創(chuàng)建和維護的代碼大大減少。
本文中的代碼實例也表達了同樣的意思。換句話說,F(xiàn)astAPI 包括許多必要的功能。它還遵循嚴格的標準,使你的代碼可以生產并更容易維護。FastAPI 的文檔也非常完善。
雖然 FastAPI 可能不像 Flask 那樣久經考驗,但越來越多的開發(fā)人員正在轉向它來提供機器學習模型或開發(fā) RESTful API。切換到 FastAPI 是一個不錯的選擇。
# 官方文檔
FastAPI:https://fastapi.tiangolo.com/
Flask:https://flask.palletsprojects.com/en/2.0.x/
# 其他資源
Porting Flask to FastAPI for ML Model Serving:https://www.pluralsight.com/tech-blog/porting-flask-to-fastapi-for-ml-model-serving/
Why we switched from Flask to FastAPI for production machine learning:https://towardsdatascience.com/why-we-switched-from-flask-to-fastapi-for-production-machine-learning-765aab9b3679
Awesome Flask:https://github.com/mjhea0/awesome-flask
Awesome FastAPI:https://github.com/mjhea0/awesome-fastapi
- EOF -
星球服務
知識星球是一個面向 全體學生和在職人員 的技術交流平臺,旨在為大家提供社招/校招準備攻略、面試題庫、面試經驗、學習路線、求職答疑、項目實戰(zhàn)案例、內推機會等內容,幫你快速成長、告別迷茫。
涉及Python,數(shù)據(jù)分析,數(shù)據(jù)挖掘,機器學習,深度學習,大數(shù)據(jù),搜光推、自然語言處理、計算機視覺、web 開發(fā)、大模型、多模態(tài)、Langchain、擴散模型、知識圖譜等方向。
我們會不定期開展知識星球立減優(yōu)惠活動,加入星球前可以添加城哥微信:dkl88191,咨詢詳情。
技術學習資料如下,星球成員可免費獲取2個,非星球成員,添加城哥微信:dkl88191,可以單獨購買。
