Xây dựng API với FastAPI và Python

FastAPI là một framework hiện đại, nhanh (hiệu suất cao), dễ học để xây dựng API với Python 3.6+ dựa trên các tiêu chuẩn API mở. Trong bài viết này, chúng ta sẽ tìm hiểu cách xây dựng một RESTful API đơn giản sử dụng FastAPI.

Tại sao chọn FastAPI?

FastAPI có nhiều ưu điểm nổi bật:

Nhanh: Hiệu suất rất cao, ngang với NodeJS và Go
Nhanh để code: Tăng tốc độ phát triển khoảng 200% đến 300%
Ít lỗi: Giảm khoảng 40% lỗi do con người
Trực quan: Hỗ trợ editor với autocompletion mạnh mẽ
Dễ học: Thiết kế để dễ sử dụng và học hỏi
Ngắn gọn: Giảm thiểu code trùng lặp
Mạnh mẽ: Có tính năng tự động tạo tài liệu API
Dựa trên tiêu chuẩn: Dựa trên và hoàn toàn tương thích với OpenAPI và JSON Schema

FastAPI - Framework hiện đại cho Python

Cài đặt FastAPI

Để bắt đầu, chúng ta cần cài đặt FastAPI và Uvicorn (ASGI server):

pip install fastapi uvicorn

Tạo API đầu tiên

Hãy tạo một file main.py với nội dung sau:

from fastapi import FastAPI

app = FastAPI()

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

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

Để chạy ứng dụng, sử dụng lệnh:

uvicorn main:app --reload

Bây giờ, API của bạn đã chạy tại http://127.0.0.1:8000. Bạn có thể truy cập:

Tạo mô hình dữ liệu với Pydantic

FastAPI sử dụng Pydantic để xác thực dữ liệu, serialize và deserialize. Hãy tạo một mô hình Pydantic:

from fastapi import FastAPI
from pydantic import BaseModel
from typing import Optional

app = FastAPI()

class Item(BaseModel):
    name: str
    price: float
    is_offer: Optional[bool] = None

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

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

@app.put("/items/{item_id}")
def update_item(item_id: int, item: Item):
    return {"item_name": item.name, "item_id": item_id}

Xử lý HTTP Methods

FastAPI hỗ trợ tất cả các HTTP methods phổ biến:

@app.post("/items/")
def create_item(item: Item):
    return item

@app.get("/items/")
def read_items():
    return [{"name": "Item 1"}, {"name": "Item 2"}]

@app.put("/items/{item_id}")
def update_item(item_id: int, item: Item):
    return {"item_name": item.name, "item_id": item_id}

@app.delete("/items/{item_id}")
def delete_item(item_id: int):
    return {"deleted": item_id}

Xác thực và phân quyền

FastAPI cung cấp nhiều cách để xác thực người dùng. Dưới đây là một ví dụ đơn giản về xác thực với OAuth2 và JWT:

from fastapi import Depends, FastAPI, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from pydantic import BaseModel
from typing import Optional
import jwt
from datetime import datetime, timedelta

# Cài đặt bảo mật
SECRET_KEY = "your-secret-key"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

# Mô hình dữ liệu
class Token(BaseModel):
    access_token: str
    token_type: str

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

class UserInDB(User):
    hashed_password: str

# Giả lập database người dùng
fake_users_db = {
    "johndoe": {
        "username": "johndoe",
        "full_name": "John Doe",
        "email": "johndoe@example.com",
        "hashed_password": "fakehashedsecret",
        "disabled": False,
    }
}

app = FastAPI()

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

# Hàm tiện ích
def verify_password(plain_password, hashed_password):
    return plain_password + "notreallyhashed" == hashed_password

def get_user(db, username: str):
    if username in db:
        user_dict = db[username]
        return UserInDB(**user_dict)

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: Optional[timedelta] = 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
    except jwt.PyJWTError:
        raise credentials_exception
    user = get_user(fake_users_db, username=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

# Endpoints
@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

Kết nối với cơ sở dữ liệu

FastAPI không có ORM tích hợp, nhưng bạn có thể sử dụng bất kỳ thư viện Python nào như SQLAlchemy, Tortoise ORM, hoặc thậm chí là MongoDB với Motor. Dưới đây là ví dụ với SQLAlchemy:

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

# Cấu hình SQLAlchemy
SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"
engine = create_engine(SQLALCHEMY_DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

# Mô hình SQLAlchemy
class ItemDB(Base):
    __tablename__ = "items"

    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, index=True)
    description = Column(String)
    price = Column(Integer)

# Tạo bảng
Base.metadata.create_all(bind=engine)

# Mô hình Pydantic
from pydantic import BaseModel

class ItemBase(BaseModel):
    name: str
    description: str = None
    price: float

class ItemCreate(ItemBase):
    pass

class Item(ItemBase):
    id: int

    class Config:
        orm_mode = True

# Dependency
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

app = FastAPI()

@app.post("/items/", response_model=Item)
def create_item(item: ItemCreate, db: Session = Depends(get_db)):
    db_item = ItemDB(**item.dict())
    db.add(db_item)
    db.commit()
    db.refresh(db_item)
    return db_item

@app.get("/items/", response_model=list[Item])
def read_items(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    items = db.query(ItemDB).offset(skip).limit(limit).all()
    return items

@app.get("/items/{item_id}", response_model=Item)
def read_item(item_id: int, db: Session = Depends(get_db)):
    db_item = db.query(ItemDB).filter(ItemDB.id == item_id).first()
    if db_item is None:
        raise HTTPException(status_code=404, detail="Item not found")
    return db_item

Xử lý tệp tin

FastAPI cũng hỗ trợ tải lên tệp tin:

from fastapi import FastAPI, File, UploadFile
from fastapi.responses import HTMLResponse
import shutil
from typing import List

app = FastAPI()

@app.post("/uploadfile/")
async def create_upload_file(file: UploadFile):
    # Lưu tệp tin tải lên
    with open(f"uploaded_{file.filename}", "wb") as buffer:
        shutil.copyfileobj(file.file, buffer)
    
    return {"filename": file.filename}

@app.post("/uploadfiles/")
async def create_upload_files(files: List[UploadFile] = File(...)):
    return {"filenames": [file.filename for file in files]}

@app.get("/")
async def main():
    content = """
    
    
    
    
    
    
    
    
    
    
    """
    return HTMLResponse(content=content)

Xử lý lỗi và ngoại lệ

FastAPI cho phép bạn xử lý lỗi và ngoại lệ một cách dễ dàng:

from fastapi import FastAPI, HTTPException, Request
from fastapi.responses import JSONResponse

app = FastAPI()

class UnicornException(Exception):
    def __init__(self, name: str):
        self.name = name

@app.exception_handler(UnicornException)
async def unicorn_exception_handler(request: Request, exc: UnicornException):
    return JSONResponse(
        status_code=418,
        content={"message": f"Oops! {exc.name} did something. There goes a rainbow..."},
    )

@app.get("/unicorns/{name}")
async def read_unicorn(name: str):
    if name == "yolo":
        raise UnicornException(name=name)
    return {"unicorn_name": name}

@app.get("/items/{item_id}")
async def read_item(item_id: int):
    if item_id == 42:
        raise HTTPException(status_code=404, detail="Item not found")
    return {"item_id": item_id}

Middleware

FastAPI hỗ trợ middleware để xử lý các yêu cầu trước khi chúng đến endpoint:

import time
from fastapi import FastAPI, Request

app = FastAPI()

@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
    start_time = time.time()
    response = await call_next(request)
    process_time = time.time() - start_time
    response.headers["X-Process-Time"] = str(process_time)
    return response

Tạo tài liệu API

FastAPI tự động tạo tài liệu API dựa trên OpenAPI. Bạn có thể tùy chỉnh tài liệu này:

from fastapi import FastAPI

description = """
ChimichangApp API helps you do awesome stuff. 🚀

## Items

You can **read items**.

## Users

You will be able to:

* **Create users**
* **Read users**
"""

app = FastAPI(
    title="ChimichangApp",
    description=description,
    version="0.0.1",
    terms_of_service="http://example.com/terms/",
    contact={
        "name": "Deadpoolio the Amazing",
        "url": "http://x-force.example.com/contact/",
        "email": "dp@x-force.example.com",
    },
    license_info={
        "name": "Apache 2.0",
        "url": "https://www.apache.org/licenses/LICENSE-2.0.html",
    },
)

Triển khai FastAPI

Để triển khai ứng dụng FastAPI trong môi trường sản xuất, bạn có thể sử dụng Uvicorn với Gunicorn:

pip install gunicorn uvicorn

Và chạy với lệnh:

gunicorn main:app -w 4 -k uvicorn.workers.UvicornWorker

Hoặc bạn có thể đóng gói ứng dụng trong Docker:

FROM python:3.9

WORKDIR /code

COPY ./requirements.txt /code/requirements.txt

RUN pip install --no-cache-dir --upgrade -r /code/requirements.txt

COPY ./app /code/app

CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "80"]

Kết luận

FastAPI là một framework mạnh mẽ và hiện đại để xây dựng API với Python. Với hiệu suất cao, tài liệu tự động, và tính năng xác thực kiểu dữ liệu, FastAPI là một lựa chọn tuyệt vời cho các dự án API mới.

Trong bài viết này, chúng ta đã tìm hiểu các khái niệm cơ bản của FastAPI và cách xây dựng một RESTful API đơn giản. Bạn có thể tìm hiểu thêm trong tài liệu chính thức của FastAPI.

Tài nguyên học tập

Xây dựng API với FastAPI và Python

Tác giả

Bài viết liên quan

ETL từ SQL Server sang Excel với Python

Phát triển ứng dụng web với Django REST Framework

Lập trình bất đồng bộ trong Python với asyncio