Python FastAPI

Python FastAPI 5: Tính năng nâng cơ bản của FastAPI

Chào mừng bạn đến với Bài 5 của khoá học Python FastAPI. Sau 4 bài nền tảng về Python và HTTP, chúng ta sẽ chính thức xắn tay áo và đi sâu vào các tính năng cơ bản của FastAPI.

Mục tiêu của bài này là đưa bạn từ “chạy được một endpoint Hello World” đến “có thể xây dựng một REST API hoàn chỉnh, có cấu trúc, validation, và documentation”. Sau bài này, bạn sẽ đủ kỹ năng để tự xây dựng một API nho nhỏ cho dự án của mình.

Từ bài này trở đi, chúng ta sẽ xây dựng dần dần một ứng dụng thực tế: TaskAPI — một API quản lý công việc đơn giản. Ứng dụng này sẽ được mở rộng xuyên suốt các bài còn lại, từ cơ bản đến database, authentication, testing, và deploy. Bắt đầu thôi! 💪

1. Cài đặt môi trường chuẩn

Trước tiên, hãy tạo một project mới với cấu trúc gọn gàng:

# Tạo thư mục project
mkdir taskapi && cd taskapi

# Tạo và kích hoạt virtual environment
python -m venv venv
source venv/bin/activate  # Linux/macOS
# venv\Scripts\activate   # Windows

# Cài các package cần thiết
pip install "fastapi[standard]" uvicorn

# Tạo file requirements.txt
pip freeze > requirements.txt

Tại sao dùng fastapi[standard]? Đây là cách cài “đầy đủ pin” — FastAPI kèm theo uvicorn, httpx (cho testing), jinja2, python-multipart, và các dependencies khác thường dùng. Từ phiên bản FastAPI 0.100+, đây là cách cài được khuyến nghị.

2. Cấu trúc project chuẩn

Một lỗi phổ biến khi mới học là nhét tất cả vào một file main.py. Nó hoạt động, nhưng sẽ trở thành mớ bòng bong khi dự án lớn lên. Hãy cấu trúc project ngay từ đầu:

taskapi/
├── app/
│   ├── __init__.py
│   ├── main.py              # Entry point
│   ├── config.py            # Cấu hình ứng dụng
│   ├── models/              # Pydantic models (schemas)
│   │   ├── __init__.py
│   │   └── task.py
│   ├── routers/             # API routes
│   │   ├── __init__.py
│   │   └── tasks.py
│   └── services/            # Business logic
│       ├── __init__.py
│       └── task_service.py
├── tests/
├── venv/
├── requirements.txt
└── README.md

Nguyên tắc phân chia:

  • models/ chứa Pydantic models — định nghĩa schema request/response.
  • routers/ chứa code xử lý HTTP — ánh xạ URL vào function.
  • services/ chứa business logic — tách biệt với chi tiết HTTP.
  • main.py chỉ để khởi tạo app và gắn các routers vào.

Tách biệt như vậy giúp code dễ test (services test được độc lập với HTTP), dễ mở rộng, và dễ chuyển đổi sau này (ví dụ: muốn thêm gRPC, chỉ cần viết router mới dùng lại services).

3. Endpoint đầu tiên

Bắt đầu với file app/main.py:

# app/main.py
from fastapi import FastAPI

app = FastAPI(
    title="TaskAPI",
    description="API quản lý công việc đơn giản",
    version="0.1.0",
)

@app.get("/")
def read_root():
    return {"message": "Chào mừng đến với TaskAPI"}

@app.get("/health")
def health_check():
    return {"status": "healthy"}

Chạy server:

# Cách mới (FastAPI CLI)
fastapi dev app/main.py

# Cách cũ (uvicorn trực tiếp)
uvicorn app.main:app --reload

Mở trình duyệt vào http://localhost:8000 để thấy kết quả, và http://localhost:8000/docs để thấy Swagger UI tự động sinh ra. Đây là một trong những thứ “thần kỳ” nhất của FastAPI — tài liệu tương tác miễn phí, không cần viết dòng nào.

4. Path Parameters

Path parameter là phần động trong URL, khai báo bằng dấu ngoặc nhọn:

from fastapi import FastAPI

app = FastAPI()

@app.get("/tasks/{task_id}")
def get_task(task_id: int):
    return {"task_id": task_id}

# Gọi: GET /tasks/42
# Kết quả: {"task_id": 42}

Điểm quan trọng: type hint không chỉ để đẹp. FastAPI dùng nó để:

Tự động convert giá trị từ string (URL luôn là string) sang kiểu bạn khai báo (int). Tự động validate — nếu client gọi /tasks/abc, FastAPI trả về 422 với thông báo lỗi rõ ràng. Tự động ghi vào Swagger docs.

Path parameters với Enum

Khi giá trị chỉ được nhận một vài lựa chọn cố định, dùng Enum:

from enum import Enum
from fastapi import FastAPI

app = FastAPI()

class TaskStatus(str, Enum):
    pending = "pending"
    in_progress = "in_progress"
    done = "done"

@app.get("/tasks/status/{status}")
def get_tasks_by_status(status: TaskStatus):
    return {"status": status, "message": f"Lọc task theo {status.value}"}

# GET /tasks/status/pending → OK
# GET /tasks/status/xyz → 422 với error rõ ràng

Validation cho path parameters

Dùng Path để thêm các ràng buộc chi tiết:

from fastapi import FastAPI, Path

app = FastAPI()

@app.get("/tasks/{task_id}")
def get_task(
    task_id: int = Path(
        ...,
        title="Task ID",
        description="ID của task, phải lớn hơn 0",
        gt=0,               # greater than 0
        le=1000000,         # less than or equal 1 triệu
    )
):
    return {"task_id": task_id}

5. Query Parameters

Query parameters là phần sau dấu ? trong URL. FastAPI tự động coi mọi tham số function không nằm trong path là query parameter:

from fastapi import FastAPI
from typing import Optional

app = FastAPI()

@app.get("/tasks")
def list_tasks(
    skip: int = 0,                  # required? không, có default
    limit: int = 10,                # mặc định 10
    search: Optional[str] = None,   # optional, có thể None
    completed: bool = False,        # boolean
):
    return {
        "skip": skip,
        "limit": limit,
        "search": search,
        "completed": completed,
    }

# GET /tasks?skip=20&limit=50&search=urgent&completed=true

Validation nâng cao với Query

from fastapi import FastAPI, Query
from typing import List, Optional

app = FastAPI()

@app.get("/tasks")
def list_tasks(
    # Validation cho string
    search: Optional[str] = Query(
        None,
        min_length=3,
        max_length=50,
        description="Từ khoá tìm kiếm",
    ),
    # Validation cho số
    limit: int = Query(10, ge=1, le=100),
    # Query parameter nhận nhiều giá trị
    tags: List[str] = Query([]),
    # Regex
    sort: str = Query(
        "created_at",
        pattern="^(created_at|updated_at|title)$",
    ),
):
    return {
        "search": search,
        "limit": limit,
        "tags": tags,
        "sort": sort,
    }

# GET /tasks?tags=urgent&tags=work&sort=title
# → tags = ["urgent", "work"]

6. Request Body với Pydantic

Đây là nơi FastAPI toả sáng nhất. Khi cần nhận dữ liệu JSON phức tạp trong body, bạn định nghĩa một Pydantic model:

# app/models/task.py
from datetime import datetime
from enum import Enum
from typing import Optional, List
from pydantic import BaseModel, Field

class TaskStatus(str, Enum):
    pending = "pending"
    in_progress = "in_progress"
    done = "done"

class TaskPriority(str, Enum):
    low = "low"
    medium = "medium"
    high = "high"

class TaskCreate(BaseModel):
    title: str = Field(..., min_length=1, max_length=200)
    description: Optional[str] = Field(None, max_length=1000)
    priority: TaskPriority = TaskPriority.medium
    tags: List[str] = []
    due_date: Optional[datetime] = None

Sử dụng model này trong endpoint:

# app/routers/tasks.py
from fastapi import APIRouter
from app.models.task import TaskCreate

router = APIRouter(prefix="/tasks", tags=["tasks"])

@router.post("/")
def create_task(task: TaskCreate):
    # FastAPI đã tự động:
    # 1. Parse JSON từ body
    # 2. Validate theo TaskCreate
    # 3. Trả 422 nếu không hợp lệ
    # 4. Truyền task là instance của TaskCreate
    return {
        "message": "Tạo task thành công",
        "task": task.model_dump(),
    }

Khi client gửi dữ liệu sai (ví dụ thiếu title, hoặc priority không phải giá trị hợp lệ), FastAPI tự động trả về lỗi 422 với chi tiết:

{
  "detail": [
    {
      "type": "missing",
      "loc": ["body", "title"],
      "msg": "Field required",
      "input": {...}
    }
  ]
}

7. Response Model

Một nguyên tắc thiết kế quan trọng: schema tạo dữ liệu khác với schema trả về dữ liệu. Khi tạo task, client gửi title và description. Khi đọc task, response cần có thêm id, created_at, status… Hãy tách biệt chúng:

# app/models/task.py
from datetime import datetime
from pydantic import BaseModel, Field
from typing import Optional, List

class TaskBase(BaseModel):
    """Các field chung giữa create và response"""
    title: str = Field(..., min_length=1, max_length=200)
    description: Optional[str] = None
    priority: TaskPriority = TaskPriority.medium
    tags: List[str] = []

class TaskCreate(TaskBase):
    """Schema cho request tạo task"""
    pass

class TaskUpdate(BaseModel):
    """Schema cho request update - mọi field đều optional"""
    title: Optional[str] = Field(None, min_length=1, max_length=200)
    description: Optional[str] = None
    status: Optional[TaskStatus] = None
    priority: Optional[TaskPriority] = None

class TaskResponse(TaskBase):
    """Schema cho response - có thêm id, timestamps, status"""
    id: int
    status: TaskStatus
    created_at: datetime
    updated_at: datetime

    class Config:
        from_attributes = True  # Cho phép tạo từ ORM objects

Khai báo response_model trong endpoint để FastAPI validate và filter response:

from fastapi import APIRouter, status
from typing import List
from app.models.task import TaskCreate, TaskResponse

router = APIRouter(prefix="/tasks", tags=["tasks"])

@router.post(
    "/",
    response_model=TaskResponse,
    status_code=status.HTTP_201_CREATED,
)
def create_task(task: TaskCreate):
    # Giả lập lưu DB
    new_task = {
        "id": 1,
        "created_at": "2026-04-18T10:00:00",
        "updated_at": "2026-04-18T10:00:00",
        "status": "pending",
        **task.model_dump(),
    }
    return new_task

@router.get("/", response_model=List[TaskResponse])
def list_tasks():
    return []  # TODO: implement

Lợi ích của response_model:

  • Filter output: nếu bạn trả về dict có thêm field như password hay internal_notes mà không khai báo trong TaskResponse, chúng sẽ bị loại bỏ tự động — bảo mật mặc định.
  • Validate output: đảm bảo không trả về dữ liệu sai format.
  • Documentation: Swagger UI hiển thị schema chính xác của response.

8. APIRouter — Tổ chức route theo module

Khi app lớn lên, bạn không thể nhét tất cả route vào main.py. Dùng APIRouter để chia nhỏ:

# app/routers/tasks.py
from fastapi import APIRouter, HTTPException, status
from typing import List
from app.models.task import TaskCreate, TaskUpdate, TaskResponse

router = APIRouter(
    prefix="/tasks",
    tags=["tasks"],  # Nhóm trong Swagger docs
    responses={404: {"description": "Không tìm thấy task"}},
)

# Giả lập database bằng dict
_tasks_db: dict = {}
_next_id = 1

@router.get("/", response_model=List[TaskResponse])
def list_tasks(skip: int = 0, limit: int = 10):
    tasks = list(_tasks_db.values())
    return tasks[skip : skip + limit]

@router.get("/{task_id}", response_model=TaskResponse)
def get_task(task_id: int):
    task = _tasks_db.get(task_id)
    if not task:
        raise HTTPException(
            status_code=404,
            detail=f"Task {task_id} không tồn tại",
        )
    return task

@router.post(
    "/",
    response_model=TaskResponse,
    status_code=status.HTTP_201_CREATED,
)
def create_task(task: TaskCreate):
    global _next_id
    from datetime import datetime
    now = datetime.utcnow()
    new_task = {
        "id": _next_id,
        "status": "pending",
        "created_at": now,
        "updated_at": now,
        **task.model_dump(),
    }
    _tasks_db[_next_id] = new_task
    _next_id += 1
    return new_task

@router.patch("/{task_id}", response_model=TaskResponse)
def update_task(task_id: int, updates: TaskUpdate):
    task = _tasks_db.get(task_id)
    if not task:
        raise HTTPException(status_code=404, detail="Task không tồn tại")

    update_data = updates.model_dump(exclude_unset=True)
    task.update(update_data)
    from datetime import datetime
    task["updated_at"] = datetime.utcnow()
    return task

@router.delete("/{task_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_task(task_id: int):
    if task_id not in _tasks_db:
        raise HTTPException(status_code=404, detail="Task không tồn tại")
    del _tasks_db[task_id]
    return None

Gắn router vào app trong main.py:

# app/main.py
from fastapi import FastAPI
from app.routers import tasks

app = FastAPI(
    title="TaskAPI",
    description="API quản lý công việc",
    version="0.1.0",
)

app.include_router(tasks.router, prefix="/api/v1")

@app.get("/")
def root():
    return {"message": "TaskAPI v0.1.0"}

Chú ý cách nest prefix: route /tasks trong router + prefix /api/v1 ở include → URL cuối cùng là /api/v1/tasks. Cách này cho phép versioning dễ dàng.

9. Xử lý lỗi với HTTPException

Khi có lỗi, không bao giờ return None hay chuỗi lỗi. Luôn raise HTTPException:

from fastapi import HTTPException, status

@router.get("/{task_id}", response_model=TaskResponse)
def get_task(task_id: int):
    task = _tasks_db.get(task_id)
    if not task:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"Task {task_id} không tồn tại",
        )
    return task

@router.post("/")
def create_task(task: TaskCreate):
    # Kiểm tra business rule
    if any(t["title"] == task.title for t in _tasks_db.values()):
        raise HTTPException(
            status_code=status.HTTP_409_CONFLICT,
            detail="Task với title này đã tồn tại",
            headers={"X-Error-Code": "DUPLICATE_TITLE"},
        )
    # ... tạo task

Chúng ta sẽ đi sâu vào xử lý lỗi — bao gồm custom exception handlers và format lỗi thống nhất — ở Bài 8.

10. Swagger UI và ReDoc

FastAPI tự động tạo hai loại tài liệu:

http://localhost:8000/docs — Swagger UI, có thể thử gọi API ngay trên trình duyệt.

http://localhost:8000/redoc — ReDoc, giao diện đọc tài liệu đẹp hơn nhưng không thử gọi được.

Bạn có thể cải thiện chất lượng tài liệu bằng cách thêm docstring và metadata:

@router.post(
    "/",
    response_model=TaskResponse,
    status_code=status.HTTP_201_CREATED,
    summary="Tạo task mới",
    description="Tạo một task mới với các thông tin cơ bản",
    response_description="Task vừa được tạo",
)
def create_task(task: TaskCreate):
    """
    Tạo task với các thông tin:

    - **title**: Tiêu đề (bắt buộc, 1-200 ký tự)
    - **description**: Mô tả chi tiết (tuỳ chọn)
    - **priority**: Mức ưu tiên (low/medium/high)
    - **tags**: Danh sách tags
    """
    # ... implementation

Chúng ta sẽ đi sâu hơn về tài liệu API (OpenAPI/Swagger customization) ở Bài 14.

11. Example payloads trong Swagger

Thêm ví dụ vào model để Swagger UI hiển thị mẫu payload — rất tiện cho team frontend:

from pydantic import BaseModel, Field, ConfigDict

class TaskCreate(BaseModel):
    model_config = ConfigDict(
        json_schema_extra={
            "examples": [
                {
                    "title": "Hoàn thành báo cáo tuần",
                    "description": "Tổng hợp số liệu tuần 16 và gửi sếp",
                    "priority": "high",
                    "tags": ["work", "urgent"],
                }
            ]
        }
    )

    title: str = Field(..., min_length=1, max_length=200)
    description: Optional[str] = None
    priority: TaskPriority = TaskPriority.medium
    tags: List[str] = []

12. Cấu hình với Settings

Cuối cùng, đừng hardcode cấu hình. Dùng pydantic-settings để quản lý config từ environment variables:

pip install pydantic-settings
# app/config.py
from pydantic_settings import BaseSettings, SettingsConfigDict

class Settings(BaseSettings):
    model_config = SettingsConfigDict(env_file=".env")

    app_name: str = "TaskAPI"
    debug: bool = False
    database_url: str = "sqlite:///./tasks.db"
    secret_key: str = "change-me-in-production"
    cors_origins: list[str] = ["http://localhost:3000"]

settings = Settings()
# .env
DEBUG=true
DATABASE_URL=postgresql://user:pass@localhost/taskapi
SECRET_KEY=your-real-secret-key-here

Dùng settings trong app:

# app/main.py
from fastapi import FastAPI
from app.config import settings
from app.routers import tasks

app = FastAPI(
    title=settings.app_name,
    debug=settings.debug,
)

app.include_router(tasks.router, prefix="/api/v1")

Nhớ thêm .env vào .gitignore — không bao giờ commit file chứa secret vào git!

13. Code hoàn chỉnh của bài này

Sau bài học, cấu trúc project của bạn sẽ trông như thế này:

taskapi/
├── app/
│   ├── __init__.py
│   ├── main.py
│   ├── config.py
│   ├── models/
│   │   ├── __init__.py
│   │   └── task.py
│   └── routers/
│       ├── __init__.py
│       └── tasks.py
├── .env
├── .gitignore
├── requirements.txt
└── README.md

Chạy server với fastapi dev app/main.py, mở http://localhost:8000/docs, và thử tạo/đọc/cập nhật/xoá task. Bạn sẽ thấy:

Mọi endpoint đều có documentation tự động. Validation xảy ra tự động — thử gửi title rỗng hoặc priority sai, Swagger UI sẽ hiển thị lỗi rõ ràng. Response được filter đúng theo response_model.

Tất cả điều này với lượng code rất ít — đó chính là sức mạnh của FastAPI.

Bài tập

Để củng cố kiến thức, hãy thử:

Thêm query parameter status vào endpoint GET /tasks để lọc theo trạng thái. Thêm endpoint GET /tasks/{task_id}/subtasks giả lập danh sách subtask. Tạo model TaskSummary chỉ có id, title, status — dùng cho response của GET /tasks (list view không cần chi tiết). Thêm validation: due_date không được là ngày trong quá khứ. Gợi ý: dùng @field_validator của Pydantic.

Tổng kết

Trong bài này, bạn đã học được những tính năng cơ bản nhưng thiết yếu của FastAPI:

Cách thiết lập môi trường và cấu trúc project chuyên nghiệp. Path parameters, query parameters, và request body — ba cách truyền dữ liệu chính. Pydantic models cho validation tự động, và tách biệt schema create/update/response. APIRouter để tổ chức route theo module. HTTPException cho xử lý lỗi. Swagger UI và ReDoc tự động sinh từ code. Quản lý cấu hình với pydantic-settings.

Ở Bài 6 tiếp theo, chúng ta sẽ đi sâu hơn vào Route, Request và Response Model — các kỹ thuật nâng cao như nested models, response với nhiều status codes khác nhau, response tuỳ biến (file, streaming, redirect), dependency injection cho query parameters chung, và nhiều hơn nữa.

Leave a Reply

Discover more from Bệ Phóng Việt

Subscribe now to keep reading and get access to the full archive.

Continue reading