Chuyển đến nội dung chính

第 2 課:FastAPI 的 Python 基礎知識

FastAPI 的基本 Python 回顧:類型提示、資料類別、裝飾器、上下文管理器、生成器、基本非同步/等待。使用 Poetry/uv 進行虛擬環境和依賴管理。

💻 程式設計 — 第 2 課 第 2 課:FastAPI 的 Python 基礎知識

Python FastAPI:從基礎到進階

第 1 部分:Python 基礎和 FastAPI

亞洲開發網

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 專案。