[note] FastAPI

使用 PIP 安裝 fastAPI

建立虛擬環境：

# 建立名為 fastapienv 的 virtual environment
$ python -m venv fastapienv

# 啟動 fastapienv 這個 virtual environment
$ source fastapienv/bin/activate

# 檢視這個 virtual environment 中安裝了哪些套件
(fastapienv)> pip list

# 離開
(fastapienv)> deactivate

安裝 fastAPI：

(fastapienv)> pip install fastapi
(fastapienv)> pip install "uvicorn[standard]"

如果專案中已經有 requirements.txt，則可以使用下述指令來安裝套件：

$ pip install -r requirements.txt

啟動 FastAPI Server

使用 uvicorn 可以啟動 FastAPI 的應用程式：

# 進入 virtual env
$ source fastapienv/bin/activate

# 假設專案中有一支名為 books.py 的檔案
(fastapienv)> uvicorn books:app --reload

# 如果執行檔時放在 cmd 的資料夾中
(fastapienv)> uvicorn cmd.books:app --reload

除了使用 uvicorn 指令來啟動之外，也可以直接使用 fastapi CLI：

# 啟動 development mode
(fastapienv)> fastapi dev books.py

# 啟動 production mode
(fastapienv)> fastapi run books.py

定義路由注意事項

路由的匹配是由上而下的，上面配對到的話，下面就不會處理
有沒有最後的 / 是不同的

舉例來說：

# 使用 `GET /books/foobar` 會進來（沒有最後的 /）
@app.get("/books/{book_title}")

# 永遠不會進來，因為上面已經被匹配到了
@app.get("/books/{book_author}")

# 使用 `GET /books/foobar/` 會進來（有最後的 /）
@app.get("/books/{book_author}/")

在 FastAPI 中，在 function 中所定義的變數，如果有該變數有在路由中看到，就是 path parameter，否則，就會是 query parameter。

Data Validation

使用 pydantic 中的 BaseModel 和 Field來針對 request payload 進行 field validation
使用 pydantic 中的 conint 或 constr 可以做到，某欄位是 optional，但若使用者有給值時才進行驗證
使用 fastapi 中的 Path 來針對 path parameter 進行 validation
使用 fastapi 中的 Query 來針對 query parameter 進行 validation

from typing import Optional

from pydantic import BaseModel, Field

class Book:
    id: int
    title: str
    author: str
    description: str
    rating: int

    def __init__(self, id: int, title: str, author: str, description: str, rating: int):
        self.id = id
        self.title = title
        self.author = author
        self.description = description
        self.rating = rating

# STEP1: 定義 request params 的型別，如果沒加 Optional 則是 Required
# STEP2: 使用 Field 可以定義 field validation rule
class BookRequest(BaseModel):
  # 把 ID 標註成 Optional
    id: Optional[int] = Field(description='ID is not needed on create', default=None)
    title: str = Field(min_length=1)
    author: str = Field(min_length=1)
    description: str = Field(min_length=1, max_length=100)
    rating: int = Field(gt=0, lt=6)


@app.post("/books")
# STEP2: 使用 request params 的型別
async def create_book(book_request: BookRequest):
    # 把 Book 從 BookRequest type 轉成 Book type
    new_book = Book(**book_request.model_dump())
    BOOKS.append(book_request)

Swagger

Request Payload

在 Swagger 中定義 Request Payload 的型別，並提供範例：

from pydantic import BaseModel


class UserRequest(BaseModel):
    email: str
    username: str
    first_name: str
    last_name: str
    password: str
    role: str

    # STEP 2：提供 swagger 中的 example request payload
    model_config = {
        "json_schema_extra": {
            "example": {
                "email": "pjchender@gmail.com",
                "username": "pjchender",
                "first_name": "PJ",
                "last_name": "Chen",
                "password": "password",
                "role": "admin",
            }
        }
    }



# STEP 1：在 user_request 中定義型別為 UserRequest，swagger 就會產生對應的 schema
@router.post("/users/", status_code=status.HTTP_201_CREATED)
async def create_user(db: db_dependency, user_request: UserRequest):
    # ...

    return UserResponse(
        #...
    )

Response Payload

在 Swagger 中定義 Response Payload 的型別，並提供範例：

from pydantic import BaseModel


class UserResponse(BaseModel):
    email: str
    username: str
    first_name: str
    last_name: str
    role: str
    is_active: bool

    # STEP 2：定義 example response payload
    model_config = {
        "json_schema_extra": {
            "example": {
                "email": "pjchender@gmail.com",
                "username": "pjchender",
                "first_name": "PJ",
                "last_name": "Chen",
                "role": "admin",
                "is_active": "true",
            }
        }
    }

# STEP 1-1：使用 response_model 讓 fastAPI 知道要根據這個作為 response body 的 schema
@router.get("/users/{user_id}/", response_model=UserResponse)
# STEP 1-2：使用 -> UserResponse（optional）
async def get_user(db: db_dependency, user_id: int) -> UserResponse:
    user = db.query(Users).get(user_id)

    if user is None:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="User not found")

    return UserResponse(
        email=user.email,
        username=user.username,
        first_name=user.first_name,
        last_name=user.last_name,
        role=user.role,
        is_active=user.is_active,
    )

在上面這兩個例子中，如果沒有加上 Step 2，則只會把型別當成 value 填入，像是這樣：

但如果有加上 Step 2，使用 model_config 明確提供 json_schema_extra 的話，則可以根據提供的範例顯示：

使用 Tag 分組

如果希望把不同的路由在 Swagger 文件中分組，可以使用 tags：

router = APIRouter(tags=["auth"])

# 不只是想做分類，連 path 都想加同一個 prefix
router = APIRouter(prefix="/auth", tags=["auth"])

如此，後面使用這裡的 router 所定義的路由，都會歸類在 auth 中：

Nullable Field

在 Swagger 中如果要讓某個欄位明確是可以填入 null 的話，要用 Field(nullable=True) 的設定：

from pydantic import BaseModel, Field, root_validator, validator

class Model(BaseModel):
  campaign_name: str | None = Field(min_length=1)
  campaign_end_time: datetime | None = Field(nullable=True)

資訊

因為在 Swagger 中 null 並不是一個型別，所以並不會有 null type，而是只能設成 nullable。

Handle Credentials

Hashed Password

安裝套件

$ pip install passlib     # Successfully installed passlib-1.7.4
$ pip install bcrypt==4.0.1  # 需要安裝這個版本的 bcrypt 才能和 passlib 搭配使用

密碼雜湊

把密碼進行雜湊的方式：

from passlib.context import CryptContext

bcrypt_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

@router.post("/users/", status_code=status.HTTP_201_CREATED)
async def create_user(db: db_dependency, user_request: UserRequest):
    user_model = Users(
        hashed_password=bcrypt_context.hash(user_request.password),
        # ...
    )

    # ...

檢查密碼

如果要檢查使用者輸入的密碼是否吻合：

def authenticate_user(username: str, password: str, db: db_dependency):
    user = db.query(Users).filter(Users.username == username).first()

    if user is None:
        return False

    if not bcrypt_context.verify(password, user.hashed_password):
        return False

    return True

JWT

安裝套件

pip install "python-jose[cryptography]"

產生 JWT

from datetime import timedelta, datetime, timezone
from jose import jwt

# The SECRET_KEY is generated the secret by "openssl rand -hex 32"
SECRET_KEY = "a788a105758e3f88e67b6cde2e1fb02ea9424d9c29eb679c7842d31c5d40f38f"
ALGORITHM = "HS256"

def create_access_token(username: str, user_id: int, expires_delta: timedelta = timedelta(hours=24)):
    to_encode = {"sub": username, "user_id": user_id}
    expires = datetime.now(timezone.utc) + expires_delta
    to_encode.update({"exp": expires})

    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)

from typing import Annotated

from fastapi import HTTPException, Depends
from fastapi.security import OAuth2PasswordRequestForm


class Token(BaseModel):
    access_token: str
    token_type: str


@router.post("/login/")
async def login_user(form_data: Annotated[OAuth2PasswordRequestForm, Depends()], db: db_dependency) -> Token:
    user = authenticate_user(form_data.username, form_data.password, db)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED, detail="Incorrect username or password"
        )

    token = create_access_token(user.username, user.id, timedelta(hours=24))
    return Token(
        access_token=token,
        token_type="bearer",
    )

驗證 JWT

from typing import Annotated

from fastapi import APIRouter, HTTPException, Depends
from fastapi.security import OAuth2PasswordRequestForm, OAuth2PasswordBearer
from jose import jwt


oauth2_bearer = OAuth2PasswordBearer(tokenUrl="/login/")


# The SECRET_KEY is generated the secret by "openssl rand -hex 32"
SECRET_KEY = "a788a105758e3f88e67b6cde2e1fb02ea9424d9c29eb679c7842d31c5d40f38f"
ALGORITHM = "HS256"


def get_current_user(token: Annotated[str, Depends(oauth2_bearer)], db: db_dependency) -> Users:
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
    )

    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        user_id: int = payload.get("user_id")
        if user_id is None:
            raise credentials_exception
    except jwt.JWTError:
        raise credentials_exception

    user = db.query(Users).get(user_id)
    if user is None:
        raise credentials_exception

    return user

參考資料

FastAPI - The Complete Course @ Udemy

使用 PIP 安裝 fastAPI​

啟動 FastAPI Server​

定義路由注意事項​

Data Validation​

Swagger​

Request Payload​

Response Payload​

使用 Tag 分組​

Nullable Field​

Handle Credentials​

Hashed Password​

安裝套件​

密碼雜湊​

檢查密碼​

JWT​

安裝套件​

產生 JWT​

Login handler​

驗證 JWT​

參考資料​

使用 PIP 安裝 fastAPI

啟動 FastAPI Server

定義路由注意事項

Data Validation

Swagger

Request Payload

Response Payload

使用 Tag 分組

Nullable Field

Handle Credentials

Hashed Password

安裝套件

密碼雜湊

檢查密碼

JWT

安裝套件

產生 JWT

Login handler

驗證 JWT

參考資料