鍵.png)
node.js + express + docker + mysql + jwt 實(shí)現(xiàn)用戶管理restful api
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ù)器。
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)境的不同配置。兩者都支持以下模式:
有關(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"
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(如 /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ā)中,這也可以幫助完成更好的代碼完成度。
與 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}
@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),需要提供請求上下文。
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")
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):
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)該“注入”到路由本身。
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)
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)
替代文檔
Flask
Flask 有一個(gè)廣泛使用的第三方管理包,稱為 Flask-Admin,用于快速對您的模型執(zhí)行 CRUD 操作。
FastAPI
截至目前,有兩個(gè)流行的 FastAPI 擴(kuò)展用于此:
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。
查看官方文檔中的以下資源以獲取更多信息:
其他資源
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)更多信息,請查看官方文檔中的測試指南。
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
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 教程。
退一步講,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í)室
node.js + express + docker + mysql + jwt 實(shí)現(xiàn)用戶管理restful api
nodejs + mongodb 編寫 restful 風(fēng)格博客 api
表格插件wpDataTables-將 WordPress 表與 Google Sheets API 連接
手把手教你用Python和Flask創(chuàng)建REST API
使用 Django 和 Django REST 框架構(gòu)建 RESTful API:實(shí)現(xiàn) CRUD 操作
ASP.NET Web API快速入門介紹
2024年在線市場平臺的11大最佳支付解決方案
完整指南:如何在應(yīng)用程序中集成和使用ChatGPT API
選擇AI API的指南:ChatGPT、Gemini或Claude,哪一個(gè)最適合你?