1. 類型提示-FastAPI 的基礎
類型提示是 FastAPI 所依賴的 Python 最重要的功能。 FastAPI 使用類型提示自動驗證資料、產生文件並提供編輯器支援。
基本類型
# Các kiểu dữ liệu cơ bản
name: str = "FastAPI"
age: int = 5
price: float = 9.99
is_active: bool = True
# Function với type hints
def greet(name: str, age: int) -> str:
return f"Hello {name}, you are {age} years old"
# Python 3.10+ union syntax
def process(value: int | str) -> str:
return str(value)
# Optional (có thể None)
def find_user(user_id: int) -> str | None:
return None
集合類型 (Python 3.9+)
# List, Dict, Set, Tuple - dùng lowercase từ Python 3.9+
names: list[str] = ["Alice", "Bob"]
scores: dict[str, int] = {"Alice": 95, "Bob": 87}
unique_ids: set[int] = {1, 2, 3}
coordinates: tuple[float, float] = (10.5, 20.3)
# Nested types
matrix: list[list[int]] = [[1, 2], [3, 4]]
users: dict[str, list[str]] = {"admin": ["read", "write"]}
# Function với collection types
def get_names(active_only: bool = True) -> list[str]:
return ["Alice", "Bob"]
進階類型
from typing import Any, Literal, TypeAlias, TypeVar, Generic
# Any - cho phép mọi kiểu (tránh dùng khi có thể)
data: Any = "anything"
# Literal - giới hạn giá trị cụ thể
Status: TypeAlias = Literal["active", "inactive", "pending"]
def set_status(status: Status) -> None:
print(f"Status: {status}")
# TypeVar và Generic
T = TypeVar("T")
class Repository(Generic[T]):
def get(self, id: int) -> T | None:
...
def list(self) -> list[T]:
...
# Callable
from collections.abc import Callable
def apply(func: Callable[[int, int], int], a: int, b: int) -> int:
return func(a, b)
2. 資料類
Dataclasses 是 Pydantic 模型的前身,有助於建立整齊儲存資料的類別:
from dataclasses import dataclass, field
from datetime import datetime
@dataclass
class User:
name: str
email: str
age: int
is_active: bool = True
created_at: datetime = field(default_factory=datetime.now)
tags: list[str] = field(default_factory=list)
@property
def display_name(self) -> str:
return f"{self.name} ({self.email})"
# Sử dụng
user = User(name="Alice", email="[email protected]", age=30)
print(user) # User(name='Alice', email='[email protected]', age=30, ...)
# Frozen (immutable)
@dataclass(frozen=True)
class Point:
x: float
y: float
3. 裝飾器
FastAPI 大量使用裝飾器(@app.get(), @app.post(),...)。需要了解裝飾器:
import functools
import time
from collections.abc import Callable
from typing import ParamSpec, TypeVar
P = ParamSpec("P")
R = TypeVar("R")
# Decorator cơ bản
def timer(func: Callable[P, R]) -> Callable[P, R]:
@functools.wraps(func)
def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
start = time.perf_counter()
result = func(*args, **kwargs)
elapsed = time.perf_counter() - start
print(f"{func.__name__} took {elapsed:.4f}s")
return result
return wrapper
@timer
def slow_function():
time.sleep(1)
return "done"
# Decorator với tham số
def retry(max_attempts: int = 3):
def decorator(func: Callable[P, R]) -> Callable[P, R]:
@functools.wraps(func)
def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
for attempt in range(max_attempts):
try:
return func(*args, **kwargs)
except Exception as e:
if attempt == max_attempts - 1:
raise
print(f"Attempt {attempt + 1} failed: {e}")
raise RuntimeError("Unreachable")
return wrapper
return decorator
@retry(max_attempts=3)
def fetch_data():
...
4. 上下文管理器
上下文管理器在 FastAPI 中對於資料庫會話、文件處理非常重要:
from contextlib import contextmanager, asynccontextmanager
# Sync context manager
@contextmanager
def db_session():
session = create_session()
try:
yield session
session.commit()
except Exception:
session.rollback()
raise
finally:
session.close()
# Async context manager (dùng nhiều trong FastAPI)
@asynccontextmanager
async def async_db_session():
session = async_create_session()
try:
yield session
await session.commit()
except Exception:
await session.rollback()
raise
finally:
await session.close()
# Class-based context manager
class Timer:
def __enter__(self):
self.start = time.perf_counter()
return self
def __exit__(self, exc_type, exc_val, exc_tb):
self.elapsed = time.perf_counter() - self.start
print(f"Elapsed: {self.elapsed:.4f}s")
return False # Don't suppress exceptions
5. 生成器和非同步生成器
FastAPI 使用生成器進行依賴注入和流響應:
# Generator cơ bản
def fibonacci(n: int):
a, b = 0, 1
for _ in range(n):
yield a
a, b = b, a + b
# Generator expression
squares = (x ** 2 for x in range(10))
# Generator cho FastAPI dependency
def get_db():
db = SessionLocal()
try:
yield db # FastAPI sẽ inject db vào route handler
finally:
db.close()
# Async generator cho streaming
async def event_stream():
while True:
data = await get_latest_event()
yield f"data: {data}\n\n"
6. 基本異步/等待
非同步/等待是FastAPI的核心功能。正確的理解將有助於編寫有效的程式碼:
import asyncio
# Async function (coroutine)
async def fetch_user(user_id: int) -> dict:
await asyncio.sleep(1) # Giả lập I/O operation
return {"id": user_id, "name": "Alice"}
# Gọi async function
async def main():
user = await fetch_user(1)
print(user)
# Chạy concurrent tasks
async def fetch_all_users(user_ids: list[int]) -> list[dict]:
# Chạy song song - KHÔNG tuần tự
tasks = [fetch_user(uid) for uid in user_ids]
results = await asyncio.gather(*tasks)
return list(results)
# asyncio.run() - entry point
asyncio.run(main())
何時在 FastAPI 中使用非同步與同步?
from fastapi import FastAPI
app = FastAPI()
# ✅ Dùng async khi có I/O operations (database, HTTP calls, file I/O)
@app.get("/users/{user_id}")
async def get_user(user_id: int):
user = await db.fetch_user(user_id) # async database query
return user
# ✅ Dùng sync khi chỉ có CPU-bound operations
# FastAPI sẽ tự chạy trong thread pool
@app.get("/compute")
def compute_heavy():
result = heavy_cpu_computation() # sync, CPU-bound
return {"result": result}
# ❌ TRÁNH: dùng async nhưng gọi sync blocking code
@app.get("/bad")
async def bad_example():
result = requests.get("https://api.example.com") # BLOCKING trong async!
return result.json()
7. 虛擬環境與依賴管理
紫外線(建議 - 2026)
# Cài đặt uv
curl -LsSf https://astral.sh/uv/install.sh | sh
# Khởi tạo project
uv init my-fastapi-project
cd my-fastapi-project
# Thêm dependencies
uv add fastapi uvicorn[standard]
uv add sqlalchemy alembic asyncpg
# Dev dependencies
uv add --dev pytest httpx ruff mypy
# Chạy
uv run uvicorn main:app --reload
# Sync dependencies
uv sync
詩歌
# Cài đặt Poetry
pip install poetry
# Khởi tạo project
poetry new my-fastapi-project
cd my-fastapi-project
# Thêm dependencies
poetry add fastapi uvicorn[standard]
poetry add sqlalchemy alembic asyncpg
# Dev dependencies
poetry add --group dev pytest httpx ruff mypy
# Chạy
poetry run uvicorn main:app --reload
8. 專案基本結構
my-fastapi-project/
├── pyproject.toml # Project config & dependencies
├── uv.lock # Lock file (uv) hoặc poetry.lock
├── README.md
├── .env # Environment variables
├── .gitignore
├── alembic.ini # Alembic config
├── alembic/ # Database migrations
│ ├── env.py
│ └── versions/
├── app/
│ ├── __init__.py
│ ├── main.py # FastAPI app entry point
│ ├── config.py # Settings & configuration
│ ├── models/ # SQLAlchemy models
│ │ ├── __init__.py
│ │ └── user.py
│ ├── schemas/ # Pydantic schemas
│ │ ├── __init__.py
│ │ └── user.py
│ ├── api/ # Route handlers
│ │ ├── __init__.py
│ │ └── v1/
│ │ ├── __init__.py
│ │ └── users.py
│ ├── services/ # Business logic
│ │ ├── __init__.py
│ │ └── user_service.py
│ ├── repositories/ # Data access layer
│ │ ├── __init__.py
│ │ └── user_repo.py
│ └── core/ # Core utilities
│ ├── __init__.py
│ ├── database.py
│ └── security.py
└── tests/
├── __init__.py
├── conftest.py
└── test_users.py
總結
在本文中,我們回顧了 FastAPI 使用的重要 Python 功能:
- 類型提示:自動驗證和文件平台
- 資料類:Pydantic 模型的前身
- 裝飾器:FastAPI 用於路由定義的模式
- 內容管理器:管理資源(資料庫會話、文件)
- 發電機:用於依賴注入和串流媒體
- 異步/等待:FastAPI高性能的核心
下一篇文章將指導您安裝和初始化實際的 FastAPI 專案。