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

Lesson 6: SQLAlchemy 2.0 & Async Database

SQLAlchemy 2.0 ORM with declarative mapping, Relationships (1-1, 1-N, N-N), Async engine with asyncpg. Session management, Unit of Work pattern. Connect PostgreSQL and MySQL.

💻 Programming — Lesson 6 Lesson 6: SQLAlchemy 2.0 & Async Database

Python FastAPI: From Basics to Advanced

Part 2: Pydantic, Database & ORM

xdev.asia

1. SQLAlchemy 2.0 - Overview

SQLAlchemy 2.0 is a new major release with many important changes: native async support, new-style query API, and improved type hints. This is the most popular ORM in the Python ecosystem.

Install

# PostgreSQL async
uv add sqlalchemy asyncpg greenlet

# Hoặc PostgreSQL sync
uv add sqlalchemy psycopg2-binary

# MySQL async
uv add sqlalchemy aiomysql

2. Database Setup with Async Engine

# app/core/database.py
from sqlalchemy.ext.asyncio import (
    AsyncSession,
    async_sessionmaker,
    create_async_engine,
)
from sqlalchemy.orm import DeclarativeBase

from app.config import settings

# Async engine
engine = create_async_engine(
    settings.database_url,
    echo=settings.debug,        # Log SQL queries khi debug
    pool_size=20,               # Connection pool size
    max_overflow=10,            # Extra connections allowed
    pool_pre_ping=True,         # Kiểm tra connection trước khi dùng
    pool_recycle=3600,          # Recycle connections sau 1 giờ
)

# Async session factory
async_session = async_sessionmaker(
    engine,
    class_=AsyncSession,
    expire_on_commit=False,     # Giữ data sau commit
)


# Base class cho tất cả models
class Base(DeclarativeBase):
    pass


# Dependency cho FastAPI
async def get_db():
    """Provide database session cho mỗi request."""
    async with async_session() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise
        finally:
            await session.close()

3. Definition of Models

# app/models/user.py
from datetime import datetime

from sqlalchemy import String, Boolean, Integer, Text, ForeignKey, Table, Column
from sqlalchemy.orm import Mapped, mapped_column, relationship

from app.core.database import Base


# Association table cho many-to-many
user_roles = Table(
    "user_roles",
    Base.metadata,
    Column("user_id", Integer, ForeignKey("users.id", ondelete="CASCADE"), primary_key=True),
    Column("role_id", Integer, ForeignKey("roles.id", ondelete="CASCADE"), primary_key=True),
)


class User(Base):
    __tablename__ = "users"

    # Primary key
    id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)

    # Required fields
    name: Mapped[str] = mapped_column(String(100), nullable=False)
    email: Mapped[str] = mapped_column(String(255), unique=True, nullable=False, index=True)
    hashed_password: Mapped[str] = mapped_column(String(255), nullable=False)

    # Optional fields
    bio: Mapped[str | None] = mapped_column(Text, nullable=True)
    avatar_url: Mapped[str | None] = mapped_column(String(500), nullable=True)
    is_active: Mapped[bool] = mapped_column(Boolean, default=True)

    # Timestamps
    created_at: Mapped[datetime] = mapped_column(default=datetime.utcnow)
    updated_at: Mapped[datetime | None] = mapped_column(
        default=None, onupdate=datetime.utcnow
    )

    # Relationships
    posts: Mapped[list["Post"]] = relationship(
        back_populates="author",
        cascade="all, delete-orphan",
        lazy="selectin",  # Eager loading
    )
    profile: Mapped["UserProfile | None"] = relationship(
        back_populates="user",
        uselist=False,  # One-to-one
        cascade="all, delete-orphan",
    )
    roles: Mapped[list["Role"]] = relationship(
        secondary=user_roles,
        back_populates="users",
        lazy="selectin",
    )

    def __repr__(self) -> str:
        return f"User(id={self.id}, name={self.name!r}, email={self.email!r})"
# app/models/post.py
from datetime import datetime
from enum import Enum as PyEnum

from sqlalchemy import String, Text, Integer, ForeignKey, Enum
from sqlalchemy.orm import Mapped, mapped_column, relationship

from app.core.database import Base


class PostStatus(str, PyEnum):
    draft = "draft"
    published = "published"
    archived = "archived"


class Post(Base):
    __tablename__ = "posts"

    id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
    title: Mapped[str] = mapped_column(String(200), nullable=False)
    slug: Mapped[str] = mapped_column(String(250), unique=True, nullable=False, index=True)
    content: Mapped[str] = mapped_column(Text, nullable=False)
    status: Mapped[PostStatus] = mapped_column(
        Enum(PostStatus), default=PostStatus.draft
    )
    view_count: Mapped[int] = mapped_column(Integer, default=0)

    # Foreign key
    author_id: Mapped[int] = mapped_column(
        ForeignKey("users.id", ondelete="CASCADE"), nullable=False, index=True
    )

    # Timestamps
    created_at: Mapped[datetime] = mapped_column(default=datetime.utcnow)
    updated_at: Mapped[datetime | None] = mapped_column(
        default=None, onupdate=datetime.utcnow
    )
    published_at: Mapped[datetime | None] = mapped_column(nullable=True)

    # Relationships
    author: Mapped["User"] = relationship(back_populates="posts")
    comments: Mapped[list["Comment"]] = relationship(
        back_populates="post",
        cascade="all, delete-orphan",
    )

    def __repr__(self) -> str:
        return f"Post(id={self.id}, title={self.title!r})"


class Comment(Base):
    __tablename__ = "comments"

    id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
    content: Mapped[str] = mapped_column(Text, nullable=False)
    post_id: Mapped[int] = mapped_column(
        ForeignKey("posts.id", ondelete="CASCADE"), nullable=False
    )
    user_id: Mapped[int] = mapped_column(
        ForeignKey("users.id", ondelete="CASCADE"), nullable=False
    )
    created_at: Mapped[datetime] = mapped_column(default=datetime.utcnow)

    # Relationships
    post: Mapped["Post"] = relationship(back_populates="comments")
    user: Mapped["User"] = relationship()

4. Async Queries

# app/repositories/user_repo.py
from sqlalchemy import select, func, or_
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import selectinload

from app.models.user import User


class UserRepository:
    def __init__(self, session: AsyncSession):
        self.session = session

    async def get_by_id(self, user_id: int) -> User | None:
        """Lấy user theo ID."""
        result = await self.session.execute(
            select(User).where(User.id == user_id)
        )
        return result.scalar_one_or_none()

    async def get_by_email(self, email: str) -> User | None:
        """Lấy user theo email."""
        result = await self.session.execute(
            select(User).where(User.email == email)
        )
        return result.scalar_one_or_none()

    async def get_with_posts(self, user_id: int) -> User | None:
        """Lấy user kèm posts (eager loading)."""
        result = await self.session.execute(
            select(User)
            .options(selectinload(User.posts))
            .where(User.id == user_id)
        )
        return result.scalar_one_or_none()

    async def list_users(
        self,
        skip: int = 0,
        limit: int = 10,
        search: str | None = None,
        is_active: bool | None = None,
    ) -> tuple[list[User], int]:
        """Lấy danh sách users với pagination và filter."""
        query = select(User)

        # Filters
        if search:
            query = query.where(
                or_(
                    User.name.ilike(f"%{search}%"),
                    User.email.ilike(f"%{search}%"),
                )
            )
        if is_active is not None:
            query = query.where(User.is_active == is_active)

        # Count total
        count_query = select(func.count()).select_from(query.subquery())
        total = (await self.session.execute(count_query)).scalar() or 0

        # Pagination
        query = query.offset(skip).limit(limit).order_by(User.created_at.desc())
        result = await self.session.execute(query)
        users = list(result.scalars().all())

        return users, total

    async def create(self, user: User) -> User:
        """Tạo user mới."""
        self.session.add(user)
        await self.session.flush()  # Flush để lấy ID
        await self.session.refresh(user)  # Refresh để lấy data mới
        return user

    async def update(self, user: User, data: dict) -> User:
        """Cập nhật user."""
        for key, value in data.items():
            setattr(user, key, value)
        await self.session.flush()
        await self.session.refresh(user)
        return user

    async def delete(self, user: User) -> None:
        """Xóa user."""
        await self.session.delete(user)
        await self.session.flush()

5. Integration with FastAPI

# app/api/v1/users.py
from fastapi import APIRouter, Depends, HTTPException, Query, status
from sqlalchemy.ext.asyncio import AsyncSession

from app.core.database import get_db
from app.models.user import User
from app.repositories.user_repo import UserRepository
from app.schemas.user import UserCreate, UserResponse, UserUpdate

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


def get_user_repo(session: AsyncSession = Depends(get_db)) -> UserRepository:
    return UserRepository(session)


@router.get("/", response_model=dict)
async def list_users(
    skip: int = Query(0, ge=0),
    limit: int = Query(10, ge=1, le=100),
    search: str | None = Query(None, min_length=1),
    repo: UserRepository = Depends(get_user_repo),
):
    users, total = await repo.list_users(skip=skip, limit=limit, search=search)
    return {
        "items": users,
        "total": total,
        "skip": skip,
        "limit": limit,
    }


@router.get("/{user_id}", response_model=UserResponse)
async def get_user(
    user_id: int,
    repo: UserRepository = Depends(get_user_repo),
):
    user = await repo.get_by_id(user_id)
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    return user


@router.post("/", response_model=UserResponse, status_code=status.HTTP_201_CREATED)
async def create_user(
    user_data: UserCreate,
    repo: UserRepository = Depends(get_user_repo),
):
    # Check email unique
    existing = await repo.get_by_email(user_data.email)
    if existing:
        raise HTTPException(status_code=400, detail="Email already registered")

    user = User(
        name=user_data.name,
        email=user_data.email,
        hashed_password=hash_password(user_data.password),
    )
    user = await repo.create(user)
    return user

6. Lifespan Events - Database initialization

# app/main.py
from contextlib import asynccontextmanager

from fastapi import FastAPI

from app.core.database import engine, Base


@asynccontextmanager
async def lifespan(app: FastAPI):
    """Application lifespan - startup/shutdown events."""
    # Startup: tạo tables (chỉ dùng dev, production dùng Alembic)
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    print("Database tables created")

    yield  # Application is running

    # Shutdown: cleanup
    await engine.dispose()
    print("Database connections closed")


app = FastAPI(lifespan=lifespan)

Summary

In this lesson learned:

  • Async Engine: Connect database with asyncpg, connection pooling
  • Declarative Models: new Mapped[] and mapped_column() styles
  • Relationships: One-to-one, One-to-many, Many-to-many
  • Async Queries: select(), where(), join(), eager loading
  • Repository Pattern: Standard CRUD operations
  • FastAPI Integration: Depends() for database session

The next article will guide Alembic migrations and database seeding.