# 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 沒有內(nèi)置的開發(fā)服務(wù)器,因此需要像 Uvicorn 或 Daphne 這樣的 ASGI 服務(wù)器。

 “Hello World” 應(yīng)用

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() 中,以實(shí)現(xiàn)開發(fā)時(shí)的熱重載。

或者,您可以直接從終端啟動(dòng)服務(wù)器:

uvicorn run fastapi_code:app

熱加載模式:

uvicorn run fastapi_code:app --reload

配置

Flask 和 FastAPI 都提供了許多選項(xiàng)來處理不同環(huán)境的不同配置。兩者都支持以下模式:

  1. 環(huán)境變量
  2. 配置文件
  3. 實(shí)例文件夾
  4. 類和繼承

有關(guān)更多信息,請參閱其各自的文檔:

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)在,在你運(yùn)行服務(wù)器之前,設(shè)置適當(dāng)?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")

同樣,在運(yùn)行服務(wù)器之前,設(shè)置適當(dāng)?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 為每個(gè)方法提供單獨(dú)的裝飾器:

@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ù)的指定類似于一個(gè) f-string 表達(dá)式。此外,你還可以利用類型提示。這里,我們在運(yùn)行時(shí)告訴 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")

默認(rèn)情況下,F(xiàn)lask會(huì)在 “templates “文件夾中尋找模板。

FastAPI

你需要安裝 Jinja:

pip install jinja2

實(shí)現(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,你需要明確地定義 “模板 “文件夾。然后對于每個(gè)響應(yīng),需要提供請求上下文。

 靜態(tài)文件

Flask

默認(rèn)情況下,F(xiàn)lask 從“static”文件夾中提供靜態(tài)文件。

FastAPI

在 FastAPI 中,需要為靜態(tài)文件掛載一個(gè)文件夾:

from fastapi.staticfiles import StaticFiles

app = FastAPI()

app.mount("/static", StaticFiles(directory="static"), name="static")

異步任務(wù)

Flask

從 Flask 2.0 開始,您可以使用?async/await?創(chuàng)建異步路由處理程序:

@app.route("/")
async def home():
result = await some_async_task()
return result

有關(guān) Flask 中異步視圖的更多信息,請查看 Flask 2.0 中的異步一文。

Flask 中的異步也可以通過使用線程(并發(fā))或多處理(并行)或 Celery 或 RQ 等工具來實(shí)現(xiàn):

  1. Asynchronous Tasks with Flask and Celery:https://testdriven.io/blog/flask-and-celery/
  2. Asynchronous Tasks with Flask and Redis Queue:https://testdriven.io/blog/asynchronous-tasks-with-flask-and-redis-queue/

FastAPI

由于 FastAPI 對 asyncio 的原生支持,它極大地簡化了異步任務(wù)。要使用的話,只需在視圖函數(shù)中添加?async?關(guān)鍵字:

@app.get("/")
async def home():
result = await some_async_task()
return result

FastAPI 還具有后臺任務(wù)功能,您可以使用它來定義返回響應(yīng)后要運(yùn)行的后臺任務(wù)。這對于不需要在發(fā)送回響應(yīng)之前完成的操作很有用。

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"}

在這里,響應(yīng)將被即時(shí)發(fā)送,而不會(huì)讓用戶等待文件處理完成。

當(dāng)你需要進(jìn)行繁重的后臺計(jì)算時(shí),或者你需要一個(gè)任務(wù)隊(duì)列來管理任務(wù)(tasks)和工作者(workers)時(shí),你可能想使用Celery 而不是 BackgroundTasks。更多內(nèi)容請參考 FastAPI 和 Celery 的異步任務(wù):https://testdriven.io/blog/fastapi-and-celery/

 依賴注入

Flask

雖然你可以實(shí)現(xiàn)自己的依賴注入解決方案,但 Flask 默認(rèn)沒有真正的一流支持。相反,你需要使用一個(gè)外部包,如 flask-injector。

FastAPI

另一方面,FastAPI 具有處理依賴注入的強(qiáng)大解決方案。

例如:

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 將獲取對在應(yīng)用程序的啟動(dòng)事件處理程序中創(chuàng)建的數(shù)據(jù)庫連接的引用。 Depends 然后用于向 FastAPI 指示路由“依賴于” get_db。因此,它應(yīng)該在路由處理程序中的代碼之前執(zhí)行,并且結(jié)果應(yīng)該“注入”到路由本身。

 數(shù)據(jù)校驗(yàn)

Flask

Flask 沒有任何內(nèi)部數(shù)據(jù)驗(yàn)證支持。您可以使用功能強(qiáng)大的 Pydantic 包通過 Flask-Pydantic 進(jìn)行數(shù)據(jù)驗(yàn)證。

FastAPI

FastAPI 如此強(qiáng)大的原因之一是它支持 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"}

在這里,我們接受一個(gè)模型?Request?的輸入。該 payload 必須包含一個(gè)用戶名和密碼。

# 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"}]}

注意到這個(gè)請求。我們把密碼 passwords 作為一個(gè)鍵而不是 password 傳遞進(jìn)去。Pydantic 模型會(huì)自動(dòng)告訴用戶,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ù)雜的對象,F(xiàn)lask 開發(fā)者經(jīng)常使用 Flask-Marshmallow

FastAPI

FastAPI 自動(dòng)序列化任何返回的字典?dict?。對于更復(fù)雜和結(jié)構(gòu)化的數(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"}

在這里,我們添加了一個(gè)包含三個(gè)輸入的 Request 模型:用戶名、電子郵件和密碼。我們還定義了一個(gè)僅包含用戶名和電子郵件的 Response 模型。輸入 Request 模型處理反序列化,而輸出 Response 模型處理對象序列化。然后通過 response_model 參數(shù)將響應(yīng)模型傳遞給裝飾器。

現(xiàn)在,如果我們將請求本身作為響應(yīng)返回,Pydantic 將省略 password ,因?yàn)槲覀兌x的響應(yīng)模型不包含密碼字段。

例如:

# output
? curl -X POST 'localhost:8000/login' \
--header 'Content-Type: application/json' \
--data-raw '{"username":"testdriven.io","email":"admin@testdriven.io","password":"testdriven.io"}'

{"username":"testdriven.io","email":"admin@testdriven.io"}

中間件

中間件被用來在每個(gè)請求被視圖功能處理之前應(yīng)用邏輯。

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)建中間件的必備工具。上述中間件計(jì)算處理請求所花費(fèi)的時(shí)間。視圖函數(shù)處理請求后,計(jì)算總處理時(shí)間并將其作為響應(yīng)頭返回。

# 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

模塊化

隨著應(yīng)用程序的發(fā)展,在某些時(shí)候你會(huì)想把類似的視圖、模板、靜態(tài)文件和模型組合在一起,以幫助把應(yīng)用程序分解成更小的組件。

Flask

在 Flask 中,藍(lán)圖被用來實(shí)現(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

同時(shí),在 FastAPI 中,模塊化是通過 APIRouter 實(shí)現(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)

其他特點(diǎn)

 自動(dòng)文檔

Flask

Flask 不會(huì)自動(dòng)創(chuàng)建開箱即用的 API 文檔。然而,有幾個(gè)擴(kuò)展可以處理這個(gè)問題,比如 flask-swagger 和 Flask RESTX,但它們需要額外的設(shè)置。

FastAPI

默認(rèn)情況下,FastAPI 支持 OpenAPI 以及 Swagger UI 和 ReDoc。這意味著每個(gè)端點(diǎn)都自動(dòng)從與端點(diǎn)關(guān)聯(lián)的元數(shù)據(jù)中記錄下來。

此處列出了所有已注冊的端點(diǎn)

替代文檔

管理應(yīng)用

Flask

Flask 有一個(gè)廣泛使用的第三方管理包,稱為 Flask-Admin,用于快速對您的模型執(zhí)行 CRUD 操作。

FastAPI

截至目前,有兩個(gè)流行的 FastAPI 擴(kuò)展用于此:

  1. FastAPI Admin – 功能性管理面板,提供用于對數(shù)據(jù)執(zhí)行 CRUD 操作的用戶界面。
  2. SQLAlchemy Admin -FastAPI/Starlette 的管理面板,可與 SQLAlchemy 模型一起使用。

 身份認(rèn)證

Flask

雖然 Flask 沒有原生解決方案,但可以使用多個(gè)第三方擴(kuò)展。

FastAPI

FastAPI 通過?fastapi.security?包原生支持許多安全和身份驗(yàn)證工具。通過幾行代碼,您可以將基本的 HTTP 身份驗(yàn)證添加到您的應(yīng)用程序中:

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 標(biāo)準(zhǔn)實(shí)現(xiàn) OAuth2 和 OpenID Connect。

查看官方文檔中的以下資源以獲取更多信息:

  1. Security Intro:https://fastapi.tiangolo.com/tutorial/security/
  2. Advanced Security:https://fastapi.tiangolo.com/advanced/security/

其他資源

  1. Web Authentication Methods Compared:https://testdriven.io/blog/web-authentication-methods/
  2. Adding Social Authentication to Flask:https://testdriven.io/blog/flask-social-auth/
  3. Session-based Auth with Flask for Single Page Apps:https://testdriven.io/blog/flask-spa-auth/
  4. Securing FastAPI with JWT Token-based Authentication:https://testdriven.io/blog/fastapi-jwt-auth/

 CORS

CORS(跨源資源共享)中間件檢查請求是否來自允許的來源。如果是,則將請求傳遞給下一個(gè)中間件或視圖函數(shù)。如果不是,它會(huì)拒絕請求,并將錯(cuò)誤響應(yīng)發(fā)送回調(diào)用者。

Flask?Flask 需要一個(gè)名為 Flask-CORS 的外部包來支持 CORS:

pip install flask-cors

基本實(shí)現(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 提供了一個(gè) TestClient。有了它,你可以直接用 FastAPI 運(yùn)行 pytest。有關(guān)更多信息,請查看官方文檔中的測試指南。

部署

 生產(chǎn)服務(wù)器

Flask

Flask 默認(rèn)運(yùn)行開發(fā) WSGI(Web 服務(wù)器網(wǎng)關(guān)接口)應(yīng)用程序服務(wù)器。對于生產(chǎn)環(huán)境,您需要使用生產(chǎn)級 WSGI 應(yīng)用服務(wù)器,例如 Gunicorn、uWSGI 或 mod_wsgi

安裝 Gunicorn:

pip install gunicorn

啟動(dòng)服務(wù):

# main.py
# app = Flask(__name__)

gunicorn main:app

FastAPI

由于 FastAPI 沒有開發(fā)服務(wù)器,您將使用 Uvicorn(或 Daphne)進(jìn)行開發(fā)和生產(chǎn)。

安裝 Uvicorn:

pip install uvicorn

啟動(dòng)服務(wù):

python
# main.py
# app = FastAPI()

uvicorn main:app

您可能希望使用 Gunicorn 來管理 Uvicorn,以便同時(shí)利用并發(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 之一。要了解如何針對生產(chǎn)對其進(jìn)行全面配置,請查看使用 Postgres、Gunicorn 和 Nginx 教程對 Flask 進(jìn)行 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"]

同樣,這是一個(gè)非常簡單的配置。 FastAPI 作者提供了幾個(gè)生產(chǎn)就緒的 Dockerfile。有關(guān)更多信息,請查看官方 FastAPI 文檔以及 Dockerizing FastAPI with Postgres、Uvicorn 和 Traefik 教程。

總結(jié)

退一步講,Django 和 Flask 是兩個(gè)最流行的基于 Python 的網(wǎng)絡(luò)框架(FastAPI 是第三大流行框架)。不過它們(Django 和 Flask)的理念非常不同。Flask 比 Django 的優(yōu)勢在于 Flask 是一個(gè)微框架。程序結(jié)構(gòu)由程序員自己決定,不強(qiáng)制執(zhí)行。開發(fā)者可以在他們認(rèn)為合適的時(shí)候添加第三方擴(kuò)展來改進(jìn)他們的代碼。也就是說,通常情況下,隨著代碼庫的增長,需要一些幾乎所有網(wǎng)絡(luò)應(yīng)用都需要的通用功能。這些功能與框架的緊密結(jié)合,使得終端開發(fā)者需要自己創(chuàng)建和維護(hù)的代碼大大減少。

本文中的代碼實(shí)例也表達(dá)了同樣的意思。換句話說,FastAPI 包括許多必要的功能。它還遵循嚴(yán)格的標(biāo)準(zhǔn),使你的代碼可以生產(chǎn)并更容易維護(hù)。FastAPI 的文檔也非常完善。

雖然 FastAPI 可能不像 Flask 那樣久經(jīng)考驗(yàn),但越來越多的開發(fā)人員正在轉(zhuǎn)向它來提供機(jī)器學(xué)習(xí)模型或開發(fā) RESTful API。切換到 FastAPI 是一個(gè)不錯(cuò)的選擇。

文章轉(zhuǎn)自微信公眾號@Python見習(xí)室

上一篇:

手把手教你用Python和Flask創(chuàng)建REST API

下一篇:

簡析Python web框架FastAPI——一個(gè)比Flask和Tornada更高性能的API 框架
#你可能也喜歡這些API文章!

我們有何不同?

API服務(wù)商零注冊

多API并行試用

數(shù)據(jù)驅(qū)動(dòng)選型,提升決策效率

查看全部API→
??

熱門場景實(shí)測,選對API

#AI文本生成大模型API

對比大模型API的內(nèi)容創(chuàng)意新穎性、情感共鳴力、商業(yè)轉(zhuǎn)化潛力

25個(gè)渠道
一鍵對比試用API 限時(shí)免費(fèi)

#AI深度推理大模型API

對比大模型API的邏輯推理準(zhǔn)確性、分析深度、可視化建議合理性

10個(gè)渠道
一鍵對比試用API 限時(shí)免費(fèi)