使用 Python 的高性能應用程序 – FastAPI 教程

已發表: 2022-03-11

良好的編程語言框架可以輕鬆更快地生產出優質產品。偉大的框架甚至使整個開發體驗變得愉快。 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

這樣，我們就完成了為所有數據交換定義模式。我們現在將注意力轉向請求處理程序，這些模式將用於免費完成所有繁重的數據轉換和驗證工作。

讓用戶註冊

首先，讓我們允許用戶註冊，因為我們所有的服務都需要經過身份驗證的用戶才能訪問。我們使用上面定義的UserCreate和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 並對其進行解碼以獲取用戶憑據。我們定義了以下函數來提供此功能。

 # install PyJWT 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 令牌。

為了接收憑證，我們將使用OAuth2PasswordRequestForm ，它是 FastAPI 安全實用程序的一部分。

 @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 提供了一個基於流行的 Requests 庫的TestClient類，我們可以使用 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。玩得開心。