Một Backend mà không có Database thì cũng giống như “não cá vàng” vậy – tắt server là quên sạch!

Trong bài này, chúng ta sẽ đi vào phần “xương sống” của mọi ứng dụng: Kết nối Database. Mình sẽ hướng dẫn các bạn combo “chuẩn chỉ” nhất hiện nay cho FastAPI: PostgreSQL + SQLModel (Async).

1. Tại sao lại là PostgreSQL + Async?

• PostgreSQL: Là hệ quản trị cơ sở dữ liệu quan hệ (RDBMS) mạnh mẽ, ổn định và “vô đối” trong thế giới Open Source hiện nay.
• Async (Bất đồng bộ): FastAPI sinh ra là để chạy Hello World… à nhầm, để chạy Async. Nếu bạn dùng thư viện DB đồng bộ (như psycopg2), bạn đang “bóp nghẹt” hiệu năng của FastAPI. Chúng ta sẽ dùng asyncpg để tận dụng tối đa sức mạnh I/O của Python.

2. SQLModel: Đứa con lai hoàn hảo

Trước đây, chúng ta phải đau đầu:

• Dùng SQLAlchemy để định nghĩa bảng (Table Model).
• Dùng Pydantic để định nghĩa dữ liệu trả về (Schema Model).
• Kết quả: Code bị lặp lại 2 lần (Duplication).

SQLModel ra đời để giải quyết việc này. Nó là sự kết hợp của SQLAlchemy + Pydantic. Bạn chỉ cần khai báo Class 1 lần, dùng được cho cả 2 mục đích!

3. Cài đặt các thư viện cần thiết

pip install fastapi uvicorn sqlmodel asyncpg

4. Thiết lập kết nối (Connection Pool)

Đây là phần quan trọng nhất. Chúng ta cần tạo một AsyncEngine để quản lý các kết nối.

database.py:

from typing import AsyncGenerator
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import sessionmaker
from sqlmodel import SQLModel

# Chuỗi kết nối PostgreSQL (dùng driver asyncpg)
DATABASE_URL = "postgresql+asyncpg://user:password@localhost/dbname"

# Tạo Engine với Connection Pool
engine = create_async_engine(
    DATABASE_URL,
    echo=True, # Log câu lệnh SQL để debug (tắt khi lên Production)
    future=True,
    pool_size=20, # Số lượng kết nối duy trì
    max_overflow=0 # Không tạo thêm kết nối quá giới hạn
)

# Tạo Session Factory
async_session_maker = sessionmaker(
    engine, class_=AsyncSession, expire_on_commit=False
)

# Dependency để lấy Session (dùng trong API)
async def get_session() -> AsyncGenerator[AsyncSession, None]:
    async with async_session_maker() as session:
        yield session

# Hàm khởi tạo DB (tạo bảng)
async def init_db():
    async with engine.begin() as conn:
        # await conn.run_sync(SQLModel.metadata.drop_all) # Chỉ dùng khi dev: Xóa bảng cũ
        await conn.run_sync(SQLModel.metadata.create_all)

5. Định nghĩa Model (Bảng dữ liệu)

models.py:

from typing import Optional
from sqlmodel import Field, SQLModel

class Hero(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str = Field(index=True) # Tạo index để search cho nhanh
    secret_name: str
    age: Optional[int] = Field(default=None, index=True)

6. Viết API (CRUD cơ bản)

main.py:

from fastapi import FastAPI, Depends
from sqlmodel import select
from sqlalchemy.ext.asyncio import AsyncSession
from models import Hero
from database import get_session, init_db

app = FastAPI()

# Tạo bảng khi app khởi chạy
@app.on_event("startup")
async def on_startup():
    await init_db()

@app.post("/heroes/")
async def create_hero(hero: Hero, session: AsyncSession = Depends(get_session)):
    session.add(hero)
    await session.commit()
    await session.refresh(hero)
    return hero

@app.get("/heroes/")
async def read_heroes(session: AsyncSession = Depends(get_session)):
    statement = select(Hero)
    result = await session.execute(statement)
    heroes = result.scalars().all()
    return heroes

7. Lưu ý quan trọng khi chạy thực tế (Production)

1. Environment Variables: Đừng bao giờ hardcode password trong code. Hãy dùng .env và thư viện python-dotenv.
2. Migrations: Với các project lớn, hãy dùng Alembic để quản lý thay đổi cấu trúc DB (thêm/sửa cột) thay vì create_all.
3. Connection Pool Size: Cấu hình pool_size phù hợp với tải của hệ thống và giới hạn của Database Server. Đừng set quá cao nếu DB yếu, cũng đừng set quá thấp nếu request nhiều.

Kết luận

Với SQLModel và FastAPI, việc thao tác với Database trở nên cực kỳ Pythonic và hiện đại. Bạn có được sự an toàn của Type Hinting, tốc độ của Async, và sự tiện lợi của ORM.

Bạn muốn học cách Deploy Database chuẩn?

Trong khóa học chuyên sâu, mình sẽ hướng dẫn các bạn cách dựng PostgreSQL bằng Docker, cấu hình Replication (Master-Slave) và backup dữ liệu tự động cho hệ thống Production.