PENALGO

Giao diện

Thực hành: FastAPI Endpoints ​

🎯 Mục tiêu

🎯 Sau bài thực hành này, bạn sẽ:

  • Xây dựng CRUD API hoàn chỉnh với FastAPI
  • Sử dụng Pydantic models cho request/response validation
  • Triển khai dependency injection và error handling chuyên nghiệp

Yêu cầu ​

Bài 1: Pydantic Models và CRUD Endpoints ​

Định nghĩa models và tạo các endpoint cơ bản.

python
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field
from datetime import datetime
app = FastAPI(title="Task Manager API")
class TaskCreate(BaseModel):
    title: str = Field(..., min_length=1, max_length=200)
    description: str = Field(default="", max_length=1000)
    priority: int = Field(default=1, ge=1, le=5)
class TaskResponse(BaseModel):
    id: int
    title: str
    description: str
    priority: int
    completed: bool
    created_at: datetime
tasks_db: dict[int, dict] = {}
next_id: int = 1
@app.post("/tasks", response_model=TaskResponse, status_code=201)
def create_task(task: TaskCreate):
    # TODO: Tạo task mới, lưu vào tasks_db
    pass
@app.get("/tasks/{task_id}", response_model=TaskResponse)
def get_task(task_id: int):
    # TODO: Trả về task hoặc raise 404
    pass
@app.get("/tasks", response_model=list[TaskResponse])
def list_tasks(completed: bool | None = None, priority: int | None = None):
    # TODO: Lọc tasks theo completed và priority
    pass

Bài 2: Error Handling ​

Xử lý lỗi chuyên nghiệp với custom exception handlers.

python
from fastapi import Request
from fastapi.responses import JSONResponse
class TaskNotFoundError(Exception):
    def __init__(self, task_id: int):
        self.task_id = task_id
@app.exception_handler(TaskNotFoundError)
async def task_not_found_handler(request: Request, exc: TaskNotFoundError):
    # TODO: Trả về JSON response với status 404
    # {"error": "not_found", "message": "...", "task_id": ...}
    pass
@app.put("/tasks/{task_id}")
def update_task(task_id: int, task: TaskCreate):
    # TODO: Cập nhật task, raise TaskNotFoundError nếu không tồn tại
    pass
@app.delete("/tasks/{task_id}", status_code=204)
def delete_task(task_id: int):
    # TODO: Xóa task, raise TaskNotFoundError nếu không tồn tại
    pass

Bài 3: Dependency Injection ​

Sử dụng Depends() để tách logic authentication và pagination.

python
from fastapi import Depends, Query, Header
def get_current_user(authorization: str = Header(...)):
    # TODO: Validate token, raise 401 nếu không hợp lệ
    pass
def pagination_params(skip: int = Query(0, ge=0), limit: int = Query(20, ge=1, le=100)):
    return {"skip": skip, "limit": limit}
@app.get("/tasks/me")
def get_my_tasks(user: dict = Depends(get_current_user), pagination: dict = Depends(pagination_params)):
    # TODO: Trả về tasks của user hiện tại với pagination
    pass

Gợi ý ​

Gợi ý Bài 1
  • Field(..., min_length=1) — ... nghĩa là required
  • Dùng global next_id hoặc class variable để tự tăng ID
  • HTTPException(status_code=404, detail="Task not found") cho lỗi
Gợi ý Bài 2
  • @app.exception_handler(ExceptionClass) đăng ký custom handler
  • Trả về JSONResponse(status_code=404, content={...})
  • Custom exceptions giúp code sạch hơn if/else kiểm tra lỗi
Gợi ý Bài 3
  • Depends(function) tự động inject kết quả của function vào endpoint
  • Dependencies có thể lồng nhau: dependency A phụ thuộc dependency B
  • Dùng Header(...) để lấy HTTP headers, Query() cho query parameters

Lời giải tham khảo ​

Xem lời giải
python
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field
from datetime import datetime
app = FastAPI(title="Task Manager API")
tasks_db: dict[int, dict] = {}
next_id: int = 1
class TaskCreate(BaseModel):
    title: str = Field(..., min_length=1, max_length=200)
    description: str = Field(default="", max_length=1000)
    priority: int = Field(default=1, ge=1, le=5)
class TaskResponse(BaseModel):
    id: int
    title: str
    description: str
    priority: int
    completed: bool
    created_at: datetime
@app.post("/tasks", response_model=TaskResponse, status_code=201)
def create_task(task: TaskCreate):
    global next_id
    new_task = {"id": next_id, **task.model_dump(), "completed": False, "created_at": datetime.now()}
    tasks_db[next_id] = new_task
    next_id += 1
    return new_task
@app.get("/tasks/{task_id}", response_model=TaskResponse)
def get_task(task_id: int):
    if task_id not in tasks_db:
        raise HTTPException(status_code=404, detail="Task không tồn tại")
    return tasks_db[task_id]

📚 Bài liên quan

Professional Engineering Standards | © 2025 HPN Corporation