FastAPI 入門教程：從零開始建立你的第一個 API

目標： 學習 FastAPI 的基本概念，建立一個簡單的 API，並了解其核心特性，如自動文件。

預計時間： 約 2 小時 (包含實際操作編寫和運行程式的時間)

參考文件： FastAPI 官方文件 - First Steps (https://fastapi.tiangolo.com/tutorial/first-steps/)

教學大綱：

是什麼？為什麼用？ (約 10 分鐘)
- 什麼是 FastAPI？
- FastAPI 的優勢 (速度、易用性、標準化、自動文件)
- 與其他 Python 框架的比較簡述
環境準備與安裝 (約 15 分鐘)
- Python 版本要求
- 安裝 FastAPI 和 ASGI 服務器 (Uvicorn)
- 創建專案目錄
你的第一個 FastAPI 應用 (約 20 分鐘)
- 編寫 main.py
- 創建 FastAPI 實例
- 定義第一個路徑操作 (Path Operation - GET)
- 運行開發伺服器 (Uvicorn)
- 測試 API
自動 API 文件 (約 15 分鐘)
- 探索 /docs (Swagger UI)
- 探索 /redoc
- 理解文件是如何自動生成的
路徑參數 (Path Parameters) (約 20 分鐘)
- 如何在 URL 中定義參數
- 獲取路徑參數的值
- 類型提示 (Type Hints) 與自動數據轉換
- 錯誤處理 (當類型不匹配時)
查詢參數 (Query Parameters) (約 20 分鐘)
- 如何在 URL 中定義可選參數 (?key=value)
- 設置默認值
- 定義可選參數 (Optional Parameters)
請求體 (Request Body) 與 Pydantic (約 30 分鐘)
- 什麼是請求體？為什麼需要它？
- 引入 Pydantic
- 定義數據模型 (BaseModel)
- 在 POST 或 PUT 請求中使用 Pydantic 模型作為參數
- FastAPI 如何自動驗證數據
- 在 /docs 中測試帶有請求體的請求
總結與下一步 (約 10 分鐘)
- 回顧學到的內容
- 接下來可以學習什麼？

開始教程！

1. 是什麼？為什麼用？

什麼是 FastAPI？ FastAPI 是一個用於構建 API 的現代、快速（高性能）的 Web 框架，基於標準 Python 類型提示 (Python 3.7+)。
FastAPI 的優勢：
- 快速： 性能非常高，可媲美 Node.js 和 Go (歸功於 Starlette 和 Pydantic)。
- 開發速度快： 編碼速度提高約 200% 到 300%。
- 錯誤減少： 由於數據驗證，人為錯誤減少。
- 直觀： 設計得非常易於使用。
- 基於標準： 基於開放標準 API (OpenAPI, JSON Schema)。
- 自動文件： 自動生成交互式 API 文件 (Swagger UI 和 ReDoc)。
- 基於類型提示： 利用 Python 的類型提示進行數據驗證、序列化和文件生成。
為什麼選擇 FastAPI？ 如果你需要一個高性能、易於開發、並且能自動生成良好 API 文件的 Python 框架，FastAPI 是個非常好的選擇。

2. 環境準備與安裝

Python 版本： 確保你的 Python 版本是 3.7 或更高。可以在終端輸入 python --version 或 python3 --version 查看。
安裝： 我們需要安裝 FastAPI 本身，以及一個 ASGI (Asynchronous Server Gateway Interface) 伺服器來運行它。Uvicorn 是一個流行的選擇。

打開你的終端或命令提示符，運行以下命令：
```
pip install fastapi uvicorn[standard]
```
- pip install fastapi: 安裝 FastAPI 框架。
- pip install uvicorn[standard]: 安裝 Uvicorn 伺服器，[standard] 包含了額外的依賴，讓它功能更完善。
創建專案目錄： 在你的文件系統中創建一個新的文件夾作為專案目錄（例如：my-fastapi-app）。打開你喜歡的程式碼編輯器（如 VS Code），並打開這個文件夾。

3. 你的第一個 FastAPI 應用

現在，我們來編寫第一個 FastAPI 應用。

在你的專案目錄下，創建一個新的 Python 文件，命名為 main.py。
將以下程式碼粘貼或輸入到 main.py 中：
```
# main.py
from fastapi import FastAPI

# 創建一個 FastAPI 應用實例
# app 是你的主要交互對象
app = FastAPI()

# 定義一個路徑操作 (Path Operation)
# @app.get("/") 表示當收到對根路徑 "/" 的 GET 請求時，執行下面的函數
@app.get("/")
async def read_root():
    # 這個異步函數將處理請求並返回一個字典
    # FastAPI 會將字典轉換為 JSON 響應
    return {"message": "Hello, World!"}
```
- from fastapi import FastAPI: 從 fastapi 庫導入 FastAPI 類。
- app = FastAPI(): 創建 FastAPI 類的一個實例。app 是你應用程式的核心。
- @app.get("/"): 這是一個 Python 裝飾器 (decorator)。它告訴 FastAPI，下面定義的函數將處理發送到 URL 路徑 / 的 GET 請求。
- async def read_root():: 定義一個異步函數。在 FastAPI 中，你可以使用 async def 或 def 定義路徑操作函數。對於簡單的同步操作，def 也可以，但使用 async def 是推薦的，以便將來支持更複雜的異步操作。
- return {"message": "Hello, World!"}: 函數返回一個 Python 字典。FastAPI 會自動將這個字典轉換為 JSON 響應發送給客戶端。
運行開發伺服器 (Uvicorn)：

打開你的終端，切換到你的專案目錄 (my-fastapi-app)。運行以下命令：
```
uvicorn main:app --reload
```
- uvicorn: 運行 Uvicorn 伺服器。
- main: 指的是 main.py 文件（Python 模塊）。
- :app: 指的是在 main.py 文件中創建的 FastAPI 實例變數名 (app = FastAPI())。
- --reload: 啟用自動重載。這意味著當你修改並保存 main.py 文件時，伺服器會自動重啟，方便開發。
你會看到類似以下的輸出：
```
INFO:     Will watch for changes in these directories: ['/path/to/my-fastapi-app']
INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
INFO:     Started reloader process [xxxxx]
INFO:     Started server process [xxxxx]
INFO:     Waiting for application startup.
INFO:     Application startup complete.
```
伺服器現在正在 http://127.0.0.1:8000 上運行。
測試 API： 打開你的網頁瀏覽器，訪問 http://127.0.0.1:8000。你應該會看到頁面上顯示：
```
{"message": "Hello, World!"}
```
恭喜！你已經成功運行了第一個 FastAPI 應用！

4. 自動 API 文件

這是 FastAPI 的一大亮點。當你的應用運行時，它會根據你的程式碼自動生成交互式 API 文件。

Swagger UI: 打開瀏覽器，訪問 http://127.0.0.1:8000/docs。你會看到一個由 Swagger UI 生成的精美頁面，列出了你的 API 端點。
- 點擊 / 路徑下的 GET，然後點擊 “Try it out”，再點擊 “Execute”。你會看到伺服器響應 {"message": "Hello, World!"}。
ReDoc: 打開瀏覽器，訪問 http://127.0.0.1:8000/redoc。這是另一個不同風格的 API 文件頁面，通常用於提供給其他人閱讀。

這些文件是根據你的程式碼和類型提示自動生成的，你無需額外編寫文件。這極大地提高了開發效率和文件準確性。

5. 路徑參數 (Path Parameters)

很多時候，你需要從 URL 中獲取特定的值，例如獲取 ID 為 5 的商品。這就是路徑參數的作用。

修改 main.py 文件，添加一個新的路徑操作：

from fastapi import FastAPI

app = FastAPI()

@app.get("/")
async def read_root():
    return {"message": "Hello, World!"}

# 新增一個帶有路徑參數的路徑操作
# {item_id} 表示這是一個路徑參數
@app.get("/items/{item_id}")
# 函數參數 item_id 接收路徑參數的值
# item_id: int 表示使用類型提示，告訴 FastAPI item_id 應該是整數
async def read_item(item_id: int):
    return {"item_id": item_id}

由於你使用了 --reload 運行 Uvicorn，伺服器會自動重啟。
測試：
- 訪問 http://127.0.0.1:8000/items/5。你會看到 {"item_id": 5}。
- 訪問 http://127.0.0.1:8000/items/100。你會看到 {"item_id": 100}。
- 訪問 http://127.0.0.1:8000/items/abc。FastAPI 會自動返回一個 422 Unprocessable Entity 錯誤，因為 "abc" 無法被轉換為整數 int。這就是類型提示帶來的自動數據驗證！
在 /docs 中查看： 刷新 http://127.0.0.1:8000/docs。你會看到新增了一個 /items/{item_id} 的 GET 請求。點擊它，你可以通過 “Try it out” 輸入不同的 item_id 進行測試，包括輸入非數字來觀察錯誤響應。

6. 查詢參數 (Query Parameters)

查詢參數是 URL 中在 ? 後面的鍵值對，通常用於過濾、分頁等可選參數。

修改 main.py 文件，再次添加或修改一個路徑操作：

from fastapi import FastAPI
from typing import Union # 導入 Union 類型提示

app = FastAPI()

@app.get("/")
async def read_root():
    return {"message": "Hello, World!"}

# 修改 /items/ 路徑操作
# item_id 是路徑參數
# q 是查詢參數，其類型是 Union[str, None]，表示可以是字符串或 None
# = None 表示 q 是一個可選參數，如果 URL 中沒有 q，它的值就是 None
@app.get("/items/{item_id}")
async def read_item(item_id: int, q: Union[str, None] = None):
    if q:
        return {"item_id": item_id, "q": q}
    return {"item_id": item_id}

# 新增一個只帶查詢參數的路徑操作
# skip 和 limit 是查詢參數，都有默認值
@app.get("/items/")
async def read_items(skip: int = 0, limit: int = 10):
    return {"skip": skip, "limit": limit}

# 新增一個帶有查詢參數的路徑操作，展示如何設置查詢參數的默認值
@app.get("/users/")
async def read_users(q: str = Query(None, min_length=3, max_length=50)):
    return {"query": q}

from typing import Union: Union 用於表示變數可以是多種類型之一。Union[str, None] 意思是可以是 str 或 None。在 Python 3.10+ 中，你可以簡寫為 str | None。
q: Union[str, None] = None: 定義了一個名為 q 的查詢參數，類型為 str 或 None，默認值為 None。因為有默認值，所以它是可選的。
skip: int = 0, limit: int = 10: 定義了兩個查詢參數 skip 和 limit，類型為 int，並設置了默認值。這意味著即使不在 URL 中提供它們，請求也能成功，它們將使用默認值。
q: str = Query(None, min_length=3, max_length=50): 使用 Query 類來定義查詢參數 q。這個參數的默認值是 None，並且設置了最小長度和最大長度的驗證。

伺服器自動重啟。
測試 /items/{item_id}：
- 訪問 http://127.0.0.1:8000/items/5。輸出 {"item_id": 5}。
- 訪問 http://127.0.0.1:8000/items/5?q=somequery。輸出 {"item_id": 5, "q": "somequery"}。
測試 /items/：
- 訪問 http://127.0.0.1:8000/items/。輸出 {"skip": 0, "limit": 10} (使用了默認值)。
- 訪問 http://127.0.0.1:8000/items/?skip=20。輸出 {"skip": 20, "limit": 10}。
- 訪問 http://127.0.0.1:8000/items/?limit=50。輸出 {"skip": 0, "limit": 50}。
- 訪問 http://127.0.0.1:8000/items/?skip=20&limit=50。輸出 {"skip": 20, "limit": 50}。
- 訪問 http://127.0.0.1:8000/items/?skip=abc。同樣會得到 422 錯誤，因為 skip 要求是整數。
- 訪問 http://127.0.0.1:8000/users/?q=jo。輸出 {"query": "jo"}。
- 訪問 http://127.0.0.1:8000/users/?q=。輸出 {"query": ""}。
- 訪問 http://127.0.0.1:8000/users/?q=ab。會返回 422 錯誤，因為 q 的最小長度是 3。
在 /docs 中查看： 刷新 /docs。查看這兩個路徑操作的文檔。你會看到查詢參數被清晰地列出，包括它們的類型和默認值。在 “Try it out” 中測試不同的查詢參數組合。

7. 請求體 (Request Body) 與 Pydantic

對於 POST、PUT、DELETE 等請求，客戶端通常會發送數據給伺服器，這些數據放在請求體中（最常見的是 JSON 格式）。FastAPI 使用 Pydantic 庫來處理請求體的數據驗證、解析和序列化。

修改 main.py 文件，添加一個 POST 路徑操作：

from fastapi import FastAPI
from typing import Union
from pydantic import BaseModel # 導入 Pydantic 的 BaseModel

# 定義一個 Pydantic 模型
# 它繼承自 BaseModel
class Item(BaseModel):
    name: str
    description: Union[str, None] = None # 可選的 description 屬性
    price: float
    tax: Union[float, None] = None     # 可選的 tax 屬性

app = FastAPI()

@app.get("/")
async def read_root():
    return {"message": "Hello, World!"}

@app.get("/items/{item_id}")
async def read_item(item_id: int, q: Union[str, None] = None):
    if q:
        return {"item_id": item_id, "q": q}
    return {"item_id": item_id}

@app.get("/items/")
async def read_items(skip: int = 0, limit: int = 10):
    return {"skip": skip, "limit": limit}

# 新增一個 POST 路徑操作
# 在函數參數中聲明一個類型為 Item 的參數 item
# FastAPI 會期望請求體是 JSON，並嘗試將其解析到一個 Item 對象中
@app.post("/items/")
async def create_item(item: Item):
    # 你可以直接訪問 item 對象的屬性
    # 例如：item.name, item.price
    # 返回 item 對象，FastAPI 會將它轉換回 JSON 響應
    return item

# 再新增一個 POST 路徑操作，展示如何獲取請求體和路徑參數
@app.post("/items/{item_id}")
async def update_item(item_id: int, item: Item):
    return {"item_id": item_id, **item.model_dump()} # 使用 item.model_dump() 或 item.dict() 將 Pydantic 模型轉為字典

from pydantic import BaseModel: 導入 Pydantic 的基類，用於定義數據模型。
class Item(BaseModel): ...: 定義了一個名為 Item 的 Pydantic 模型。它有四個屬性 (name, description, price, tax)，並指定了它們的類型。FastAPI/Pydantic 會根據這些類型來驗證傳入的 JSON 數據。
- name: str: name 是必須的，且必須是字符串。
- description: Union[str, None] = None: description 是可選的，可以是字符串或 None，默認為 None。
- price: float: price 是必須的，且必須是浮點數。
- tax: Union[float, None] = None: tax 是可選的，可以是浮點數或 None，默認為 None。
async def create_item(item: Item):: 定義一個 POST 請求處理函數。參數 item: Item 告訴 FastAPI 期望請求體是一個可以匹配 Item 模型結構的 JSON 對象。FastAPI 會自動：
1. 讀取請求體。
2. 將 JSON 解析為 Python 字典。
3. 使用 Pydantic 驗證字典數據是否符合 Item 模型的定義。
4. 如果數據有效，創建一個 Item 對象並將其作為參數傳入函數。
5. 如果數據無效（例如缺少必需字段，或類型不匹配），自動返回一個 422 Unprocessable Entity 響應，並包含詳細的錯誤信息。
return item: 直接返回 item 對象。FastAPI 會自動使用 Pydantic 將這個對象轉換回 JSON 響應。
**item.model_dump(): Python 的字典解包語法。item.model_dump() (或在舊版本 Pydantic 中是 item.dict()) 將 Pydantic 模型轉換為一個字典。我們將這個字典的內容解包到一個新的字典中，並加上 item_id。

伺服器自動重啟。
在 /docs 中測試： 刷新 /docs。你會看到新增了兩個 POST 請求。
- 點擊 /items/ 的 POST 請求。你會看到它要求一個 Request body，並顯示了 Item 模型的結構（包括屬性名稱、類型、是否必須）。點擊 “Try it out”，你會看到一個預填充的請求體 JSON 範例。
- 輸入一個有效的 JSON 請求體，例如：
```
{
  "name": "Fork",
  "description": "A lovely fork",
  "price": 3.5,
  "tax": 0.5
}
```
  點擊 “Execute”。如果成功，你會收到一個 200 響應，響應體就是你發送的數據。
- 嘗試發送一個無效的請求體，例如移除 name 或 price 字段，或者將 price 設置為非數字。點擊 “Execute”。你會看到 422 錯誤響應和詳細的驗證錯誤信息。
- 測試 /items/{item_id} 的 POST 請求，它需要路徑參數和請求體。

8. 總結與下一步

恭喜！你已經完成了 FastAPI 的入門學習，掌握了以下核心概念：

安裝和運行 FastAPI 應用。
創建基本的 路徑操作 (GET, POST)。
使用 路徑參數 從 URL 中獲取值。
使用 查詢參數 處理可選值和默認值。
理解並利用 Pydantic 定義 請求體 模型，實現自動 數據驗證 和 序列化。
體驗了 FastAPI 強大的自動 API 文件 功能 (/docs, /redoc)。

你學到的這些是構建大多數 API 的基礎。

9. 進階功能探索

9.1 使用依賴注入 (Dependency Injection)

FastAPI 提供了強大的依賴注入系統，可以用於：

共享邏輯 (如身份驗證)
資料庫連接管理
參數驗證與提取

讓我們實現一個簡單的依賴項，用於參數驗證：

from fastapi import FastAPI, Depends, Query, HTTPException

app = FastAPI()

# 定義一個依賴函數
async def common_parameters(q: str = None, skip: int = 0, limit: int = 100):
    if skip < 0:
        raise HTTPException(status_code=400, detail="skip 不能為負數")
    if limit > 1000:
        raise HTTPException(status_code=400, detail="limit 不能超過1000")
    return {"q": q, "skip": skip, "limit": limit}

# 使用依賴項
@app.get("/items/")
async def read_items(commons: dict = Depends(common_parameters)):
    return {"params": commons}

@app.get("/users/")
async def read_users(commons: dict = Depends(common_parameters)):
    return {"params": commons}

9.2 文件上傳處理

FastAPI 使得文件上傳處理變得簡單：

from fastapi import FastAPI, File, UploadFile
from fastapi.responses import FileResponse
import shutil
from pathlib import Path

app = FastAPI()

@app.post("/upload-file/")
async def create_upload_file(file: UploadFile = File(...)):
    # 儲存上傳文件
    upload_dir = Path("static/uploads")
    upload_dir.mkdir(parents=True, exist_ok=True)
    
    file_path = upload_dir / file.filename
    with open(file_path, "wb") as buffer:
        shutil.copyfileobj(file.file, buffer)
    
    return {"filename": file.filename, "content_type": file.content_type}

@app.get("/files/{filename}")
async def get_file(filename: str):
    file_path = Path("static/uploads") / filename
    
    if not file_path.exists():
        raise HTTPException(status_code=404, detail="檔案不存在")
        
    return FileResponse(file_path)

9.3 資料庫整合 (使用 SQLAlchemy)

整合資料庫是大部分應用程式的必要功能：

from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, Session
from pydantic import BaseModel

# 定義資料庫模型
Base = declarative_base()

class UserDB(Base):
    __tablename__ = "users"
    
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, index=True)
    email = Column(String, unique=True, index=True)

# 創建資料庫引擎 (這裡使用SQLite)
SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"
engine = create_engine(SQLALCHEMY_DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# 創建所有表
Base.metadata.create_all(bind=engine)

# 定義 Pydantic 模型
class User(BaseModel):
    name: str
    email: str
    
    class Config:
        orm_mode = True

class UserCreate(User):
    pass
    
class UserResponse(User):
    id: int

# 定義依賴項，獲取數據庫會話
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

app = FastAPI()

@app.post("/users/", response_model=UserResponse)
def create_user(user: UserCreate, db: Session = Depends(get_db)):
    # 檢查郵箱是否已存在
    db_user = db.query(UserDB).filter(UserDB.email == user.email).first()
    if db_user:
        raise HTTPException(status_code=400, detail="Email already registered")
    
    # 創建用戶
    db_user = UserDB(**user.dict())
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return db_user

@app.get("/users/", response_model=list[UserResponse])
def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    users = db.query(UserDB).offset(skip).limit(limit).all()
    return users

@app.get("/users/{user_id}", response_model=UserResponse)
def read_user(user_id: int, db: Session = Depends(get_db)):
    user = db.query(UserDB).filter(UserDB.id == user_id).first()
    if user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return user

9.4 處理認證與授權

API 安全性是關鍵，FastAPI 提供了完整的 OAuth2 與 JWT 支援：

from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from jose import JWTError, jwt
from passlib.context import CryptContext
from datetime import datetime, timedelta
from pydantic import BaseModel
from typing import Union

# 安全配置常數
SECRET_KEY = "your-secret-key"  # 生產環境應使用安全的隨機密鑰並妥善保管
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

# 模擬用戶資料庫
fake_users_db = {
    "aaron": {
        "username": "aaron",
        "full_name": "Aaron Yu",
        "email": "aaron@example.com",
        "hashed_password": "$2b$12$EixZaYVK1fsbw1ZfbX3OXePaWxn96p36WQoeG6Lruj3vjPGga31lW",  # "secret"
        "disabled": False,
    }
}

app = FastAPI()

# 密碼處理工具
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

# 模型定義
class Token(BaseModel):
    access_token: str
    token_type: str

class TokenData(BaseModel):
    username: Union[str, None] = None

class User(BaseModel):
    username: str
    email: Union[str, None] = None
    full_name: Union[str, None] = None
    disabled: Union[bool, None] = None

class UserInDB(User):
    hashed_password: str

# 驗證密碼
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 get_user(db, username: str):
    if username in db:
        user_dict = db[username]
        return UserInDB(**user_dict)
    return None

# 驗證用戶
def authenticate_user(fake_db, username: str, password: str):
    user = get_user(fake_db, username)
    if not user:
        return False
    if not verify_password(password, user.hashed_password):
        return False
    return user

# 創建訪問令牌
def create_access_token(data: dict, expires_delta: Union[timedelta, None] = 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

# 獲取當前用戶
async def get_current_user(token: str = Depends(oauth2_scheme)):
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username: str = payload.get("sub")
        if username is None:
            raise credentials_exception
        token_data = TokenData(username=username)
    except JWTError:
        raise credentials_exception
        
    user = get_user(fake_users_db, username=token_data.username)
    if user is None:
        raise credentials_exception
    return user

# 獲取活躍用戶
async def get_current_active_user(current_user: User = Depends(get_current_user)):
    if current_user.disabled:
        raise HTTPException(status_code=400, detail="Inactive user")
    return current_user

# 登入端點
@app.post("/token", response_model=Token)
async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends()):
    user = authenticate_user(fake_users_db, form_data.username, form_data.password)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect username or password",
            headers={"WWW-Authenticate": "Bearer"},
        )
    
    access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    access_token = create_access_token(
        data={"sub": user.username}, expires_delta=access_token_expires
    )
    
    return {"access_token": access_token, "token_type": "bearer"}

# 獲取當前用戶信息
@app.get("/users/me/", response_model=User)
async def read_users_me(current_user: User = Depends(get_current_active_user)):
    return current_user

9.5 WebSocket 支援

FastAPI 還支持 WebSocket，適合即時通訊應用：

from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from typing import List

app = FastAPI()

class ConnectionManager:
    def __init__(self):
        self.active_connections: List[WebSocket] = []

    async def connect(self, websocket: WebSocket):
        await websocket.accept()
        self.active_connections.append(websocket)

    def disconnect(self, websocket: WebSocket):
        self.active_connections.remove(websocket)

    async def send_personal_message(self, message: str, websocket: WebSocket):
        await websocket.send_text(message)

    async def broadcast(self, message: str):
        for connection in self.active_connections:
            await connection.send_text(message)

manager = ConnectionManager()

@app.websocket("/ws/{client_id}")
async def websocket_endpoint(websocket: WebSocket, client_id: str):
    await manager.connect(websocket)
    try:
        while True:
            data = await websocket.receive_text()
            await manager.send_personal_message(f"You sent: {data}", websocket)
            await manager.broadcast(f"Client #{client_id} says: {data}")
    except WebSocketDisconnect:
        manager.disconnect(websocket)
        await manager.broadcast(f"Client #{client_id} left the chat")

9.6 非同步資料庫連線

FastAPI 支持完全非同步的資料庫操作：

from fastapi import FastAPI, HTTPException
import asyncpg
from pydantic import BaseModel

app = FastAPI()

class Movie(BaseModel):
    title: str
    genre: str
    release_year: int

@app.on_event("startup")
async def startup_db_client():
    app.db_pool = await asyncpg.create_pool(
        "postgresql://user:password@localhost/movies_db"
    )

@app.on_event("shutdown")
async def shutdown_db_client():
    await app.db_pool.close()

@app.post("/movies/", response_model=Movie)
async def create_movie(movie: Movie):
    query = """
        INSERT INTO movies (title, genre, release_year) 
        VALUES ($1, $2, $3) 
        RETURNING title, genre, release_year
    """
    async with app.db_pool.acquire() as conn:
        record = await conn.fetchrow(
            query, movie.title, movie.genre, movie.release_year
        )
        return dict(record)

@app.get("/movies/")
async def read_movies():
    async with app.db_pool.acquire() as conn:
        records = await conn.fetch("SELECT title, genre, release_year FROM movies")
        return [dict(record) for record in records]

接下來可以學習什麼？

官方文件提供了豐富的內容，你可以繼續深入學習：

Query Parameters and String Validations: 對查詢參數進行更詳細的驗證 (長度、正則表達式等)。
Path Parameters and Numeric Validations: 對路徑參數進行更詳細的驗證 (大小、範圍等)。
Request Body - Fields: 在 Pydantic 模型中添加字段級別的驗證和元數據。
Request Body - Nested Models: 處理更複雜的嵌套 JSON 結構。
Request Body - Declare multiple parameters: 同時接收請求體、路徑參數、查詢參數、Header、Cookie 等。
Extra Data Types: 使用 UUID, datetime, date, time, timedelta 等標準 Python 類型。
Status Codes: 如何設置 API 響應的 HTTP 狀態碼。
Forms and File Uploads: 處理表單數據和文件上傳。
Dependencies: FastAPI 中非常強大的依賴注入系統，用於共用邏輯、數據庫連接、認證等。
Security: 實現 API 認證和授權 (OAuth2, JWT 等)。
Handling Errors: 自定義錯誤處理。
SQL Databases: 如何與數據庫集成 (如 SQLAlchemy, ORM)。
Testing: 如何編寫測試你的 FastAPI 應用。
Deployment: 如何部署你的應用到生產環境。

10. 生產部署實踐

10.1 Gunicorn + Uvicorn 部署

在生產環境中，推薦使用 Gunicorn 作為進程管理器，配合 Uvicorn 工作器：

# 安裝 Gunicorn
pip install gunicorn

# 啟動伺服器
gunicorn -w 4 -k uvicorn.workers.UvicornWorker main:app

10.2 容器化部署（使用 Docker）

# 使用 Python 官方映像作為基礎
FROM python:3.9-slim

# 設置工作目錄
WORKDIR /app

# 複製依賴文件
COPY requirements.txt .

# 安裝依賴
RUN pip install --no-cache-dir -r requirements.txt

# 複製應用程式碼
COPY . .

# 暴露端口
EXPOSE 8000

# 啟動命令
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

建議你接下來按照官方文件中的「Tutorial - User Guide」順序繼續學習，逐步掌握 FastAPI 的更多高級特性。

FastAPI 入門教程：從零開始建立你的第一個 API

FastAPI 入門教程：從零開始建立你的第一個 API

9. 進階功能探索

9.1 使用依賴注入 (Dependency Injection)

9.2 文件上傳處理

9.3 資料庫整合 (使用 SQLAlchemy)

9.4 處理認證與授權

9.5 WebSocket 支援

9.6 非同步資料庫連線

10. 生產部署實踐

10.1 Gunicorn + Uvicorn 部署

10.2 容器化部署（使用 Docker）

標籤

延伸閱讀