Home >> Blog >> 使用 Python 的高性能應用程式 FastAPI 教學
使用 Python 的高性能應用程式 FastAPI 教學
良好的編寫程式語言框架可以輕鬆更快地生產出優質產品。偉大的框架甚至使整個SEO搜尋引擎優化開發體驗變得愉快。FastAPI 是一個新的 Python Web 框架，功能強大且使用起來很愉快。以下特性使 FastAPI 值得一試：
- 速度：FastAPI 是最快的 Python Web 框架之一。事實上，它的速度與 Node.js 和 Go 不相上下。檢查這些性能測試。
- 詳細且易於使用的開發人員文檔
- 鍵入提示您的程式碼並獲得免費的資料驗證和轉換。
- 使用依賴注入輕鬆創建插件。
構建一個 TODO 應用程式
為了探索 FastAPI 背後的重要理念，讓我們構建一個 TODO 應用程式，它為用戶設置待辦事項列表。我們的微型應用程式將提供以下功能：
- 註冊和登錄
- 添加新的 TODO 項
- 獲取所有 TODO 的列表
- 刪除/更新 TODO 項目
資料模型的 SQLAlchemy
我們的應用只有兩個模型：User 和 TODO。在 Python 的資料庫工具包 SQLAlchemy 的幫助下，我們可以像這樣表達我們的模型：
class User(Base):
__tablename__ = "users"
id = Column(Integer, primary_key=True, index=True)
lname = Column(String)
fname = Column(String)
email = Column(String, unique=True, index=True)
todos = relationship("TODO", back_populates="owner", cascade="all, delete-orphan")
class TODO(Base):
__tablename__ = "todos"
id = Column(Integer, primary_key=True, index=True)
text = Column(String, index=True)
completed = Column(Boolean, default=False)
owner_id = Column(Integer, ForeignKey("users.id"))
owner = relationship("User", back_populates="todos")
一旦我們的模型準備就緒，讓我們為 SQLAlchemy 編寫配置文件，以便它知道如何與資料庫建立連接。
import os
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
SQLALCHEMY_DATABASE_URL = os.environ['SQLALCHEMY_DATABASE_URL']
engine = create_engine(
SQLALCHEMY_DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()
釋放類型提示的力量
任何 API 項目的相當大一部分都涉及資料驗證和轉換等日常工作。在我們開始編寫請求處理程式之前，讓我們先解決它。使用 FastAPI，我們使用 pydantic 模型表達傳入/傳出資料的模式，然後使用這些 pydantic 模型輸入提示並享受免費的資料驗證和轉換。請注意，這些模型與我們的資料庫工作流程無關，僅指定流入和流出 REST 介面的資料的形狀。要編寫 pydantic 模型，請考慮 User 和 TODO 信息流入和流出的所有方式。
傳統上，新用戶將註冊我們的 TODO 服務，而現有用戶將登錄。這兩種交互都處理用戶信息，但資料的形式會有所不同。我們在註冊時需要更多用戶信息，在登錄時需要最少的信息（只有電子郵件和密碼）。這意味著我們需要兩個 pydantic 模型來表達這兩種不同形式的用戶信息。
然而，在我們的 TODO 應用程式中，我們將利用 FastAPI 中內置的 OAuth2 支持來實現基於 JSON Web 令牌 (JWT) 的登錄流程。我們只需要在UserCreate此處定義一個模式來指定將流入我們的註冊端點的資料和一個UserBase模式以在註冊過程成功的情況下作為響應返回。
from pydantic import BaseModel
from pydantic import EmailStr
class UserBase(BaseModel):
email: EmailStr
class UserCreate(UserBase):
lname: str
fname: str
password: str
在這裡，我們將姓氏、名字和密碼標記為字符串，但可以通過使用 pydantic約束字符串來進一步收緊它，這些字符串可以進行最小長度、最大長度和正則表達式等檢查。
為了支持 TODO 項的創建和列表，我們定義了以下模式：
class TODOCreate(BaseModel):
text: str
completed: bool
為了支持現有 TODO 項的更新，我們定義了另一個模式：
class TODOUpdate(TODOCreate):
id: int
這樣，我們就完成了為所有資料交換定義模式。我們現在將注意力轉向請求處理程式，這些模式將用於免費完成所有繁重的資料轉換和驗證工作。
讓用戶註冊
首先，讓我們允許用戶註冊，因為我們所有的服務都需要經過身份驗證的用戶才能訪問。我們使用上面定義的UserCreateand模式編寫我們的第一個請求處理程式。UserBase
@app.post("/api/users", response_model=schemas.User)
def signup(user_data: schemas.UserCreate, db: Session = Depends(get_db)):
"""add new user"""
user = crud.get_user_by_email(db, user_data.email)
if user:
raise HTTPException(status_code=409,
detail="Email already registered.")
signedup_user = crud.create_user(db, user_data)
return signedup_user
在這段簡短的程式碼中發生了很多事情。我們使用裝飾器來指定 HTTP 動詞、URI 和成功響應的模式。為了確保用戶提交了正確的資料，我們使用較早定義的UserCreate模式鍵入提示請求正文。該方法定義了另一個用於獲取資料庫句柄的參數——這是實際的依賴注入，將在本教程後面討論。
保護我們的 API
我們希望我們的應用程式具有以下安全功能：
- 密碼哈希
- 基於 JWT 的身份驗證
對於密碼哈希，我們可以使用 Passlib。讓我們定義處理密碼散列和檢查密碼是否正確的函數。
from passlib.context import CryptContext
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
def verify_password(plain_password, hashed_password):
return pwd_context.verify(plain_password, hashed_password)
def get_password_hash(password):
return pwd_context.hash(password)
def authenticate_user(db, email: str, password: str):
user = crud.get_user_by_email(db, email)
if not user:
return False
if not verify_password(password, user.hashed_password):
return False
return user
要啟用基於 JWT 的身份驗證，我們需要生成 JWT 並對其進行解碼以獲取用戶憑據。我們定義了以下函數來提供此功能。
import jwt
from fastapi.security import OAuth2PasswordBearer
SECRET_KEY = os.environ['SECRET_KEY']
ALGORITHM = os.environ['ALGORITHM']
def create_access_token(*, data: dict, expires_delta: timedelta = None):
to_encode = data.copy()
if expires_delta:
expire = datetime.utcnow() + expires_delta
else:
expire = datetime.utcnow() + timedelta(minutes=15)
to_encode.update({"exp": expire})
encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
return encoded_jwt
def decode_access_token(db, token):
credentials_exception = HTTPException(
status_code=HTTP_401_UNAUTHORIZED,
detail="Could not validate credentials",
headers={"WWW-Authenticate": "Bearer"},
)
try:
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
email: str = payload.get("sub")
if email is None:
raise credentials_exception
token_data = schemas.TokenData(email=email)
except PyJWTError:
raise credentials_exception
user = crud.get_user_by_email(db, email=token_data.email)
if user is None:
raise credentials_exception
return user
成功登錄時發出令牌
現在，我們將定義一個登錄端點並實現 OAuth2 密碼流。該端點將收到一封電子郵件和密碼。我們將根據資料庫檢查憑據，並在成功時向用戶發出 JSON Web 令牌。
為了接收憑證，我們將使用OAuth2PasswordRequestFormFastAPI 安全實用程式的一部分。
@app.post("/api/token", response_model=schemas.Token)
def login_for_access_token(db: Session = Depends(get_db),
form_data: OAuth2PasswordRequestForm = Depends()):
"""generate access token for valid credentials"""
user = authenticate_user(db, form_data.username, form_data.password)
if not user:
raise HTTPException(
status_code=HTTP_401_UNAUTHORIZED,
detail="Incorrect email or password",
headers={"WWW-Authenticate": "Bearer"},
)
access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
access_token = create_access_token(data={"sub": user.email},
expires_delta=access_token_expires)
return {"access_token": access_token, "token_type": "bearer"}
使用依賴注入訪問資料庫和保護端點
我們已經設置了登錄端點，該端點在成功登錄時向用戶提供 JWT。用戶可以將此令牌保存在本地存儲中，並將其作為授權標頭顯示給我們的後端。僅希望登錄用戶訪問的端點可以解碼令牌並找出請求者是誰。這種工作不依賴於特定的端點，而是在所有受保護的端點中使用的共享邏輯。最好將令牌解碼邏輯設置為可在任何請求處理程式中使用的依賴項。
在 FastAPI 中，我們的路徑操作函數（請求處理程式）將依賴於get_current_user. get_current_user依賴項需要與資料庫建立連接並掛鉤到 FastAPI 的邏輯OAuth2PasswordBearer以獲取令牌。我們將通過get_current_user依賴其他函數來解決這個問題。這樣，我們就可以定義依賴鏈，這是一個非常強大的概念。
def get_db():
"""provide db session to path operation functions"""
try:
db = SessionLocal()
yield db
finally:
db.close()
def get_current_user(db: Session = Depends(get_db),
token: str = Depends(oauth2_scheme)):
return decode_access_token(db, token)
@app.get("/api/me", response_model=schemas.User)
def read_logged_in_user(current_user: models.User = Depends(get_current_user)):
"""return user settings for current user"""
return current_user
登錄用戶可以 CRUD TODO
在我們為 TODO 創建、讀取、更新、刪除 (CRUD) 編寫路徑操作函數之前，我們定義了以下幫助函數來在 db 上執行實際的 CRUD。
def create_todo(db: Session, current_user: models.User, todo_data: schemas.TODOCreate):
todo = models.TODO(text=todo_data.text,
completed=todo_data.completed)
todo.owner = current_user
db.add(todo)
db.commit()
db.refresh(todo)
return todo
def update_todo(db: Session, todo_data: schemas.TODOUpdate):
todo = db.query(models.TODO).filter(models.TODO.id == id).first()
todo.text = todo_data.text
todo.completed = todo.completed
db.commit()
db.refresh(todo)
return todo
def delete_todo(db: Session, id: int):
todo = db.query(models.TODO).filter(models.TODO.id == id).first()
db.delete(todo)
db.commit()
def get_user_todos(db: Session, userid: int):
return db.query(models.TODO).filter(models.TODO.owner_id == userid).all()
這些 db 級函數將用於以下 REST 端點：
@app.get("/api/mytodos", response_model=List[schemas.TODO])
def get_own_todos(current_user: models.User = Depends(get_current_user),
db: Session = Depends(get_db)):
"""return a list of TODOs owned by current user"""
todos = crud.get_user_todos(db, current_user.id)
return todos
@app.post("/api/todos", response_model=schemas.TODO)
def add_a_todo(todo_data: schemas.TODOCreate,
current_user: models.User = Depends(get_current_user),
db: Session = Depends(get_db)):
"""add a TODO"""
todo = crud.create_meal(db, current_user, meal_data)
return todo
@app.put("/api/todos/{todo_id}", response_model=schemas.TODO)
def update_a_todo(todo_id: int,
todo_data: schemas.TODOUpdate,
current_user: models.User = Depends(get_current_user),
db: Session = Depends(get_db)):
"""update and return TODO for given id"""
todo = crud.get_todo(db, todo_id)
updated_todo = crud.update_todo(db, todo_id, todo_data)
return updated_todo
@app.delete("/api/todos/{todo_id}")
def delete_a_meal(todo_id: int,
current_user: models.User = Depends(get_current_user),
db: Session = Depends(get_db)):
"""delete TODO of given id"""
crud.delete_meal(db, todo_id)
return {"detail": "TODO Deleted"}
編寫測試
讓我們為我們的 TODO API 編寫一些測試。FastAPI 提供了一個TestClient基於流行的 Requests 庫的類，我們可以使用 Pytest 運行測試。
為了確保只有登錄用戶可以創建 TODO，我們可以這樣寫：
from starlette.testclient import TestClient
from .main import app
client = TestClient(app)
def test_unauthenticated_user_cant_create_todos(): todo=dict(text="run a mile", completed=False)
response = client.post("/api/todos", data=todo)
assert response.status_code == 401
以下測試檢查我們的登錄端點並在提供有效登錄憑據時生成 JWT。
def test_user_can_obtain_auth_token():
response = client.post("/api/token", data=good_credentials)
assert response.status_code == 200
assert 'access_token' in response.json()
assert 'token_type' in response.json()
總結一下
我們已經使用 FastAPI 實現了一個非常簡單的 TODO 應用程式。到目前為止，您已經看到類型提示的強大功能在通過我們的 REST 介面定義傳入和傳出資料的形狀時得到了很好的利用。我們在一處定義模式並將其留給 FastAPI 來應用資料驗證和轉換。另一個值得注意的特性是依賴注入。我們用這個概念封裝了獲取資料庫連接的共享邏輯，解碼JWT獲取當前登錄的用戶，實現簡單的OAuth2密碼和承載。我們還看到瞭如何將依賴項鍊接在一起。
我們可以輕鬆地應用此概念來添加基於角色的訪問等功能。此外，我們正在編寫簡潔而強大的程式碼，而無需學習框架的特性。簡而言之，FastAPI 是一組強大的工具，您無需學習，因為它們只是現代 Python。