1.SQLAlchemy 2.0 - 概述
SQLAlchemy 2.0 是一個新的主要版本，具有許多重要的變更：本機非同步支援、新型查詢 API 和改進的類型提示。這是Python生態系中最受歡迎的ORM。
安裝
# PostgreSQL async
uv add sqlalchemy asyncpg greenlet
# Hoặc PostgreSQL sync
uv add sqlalchemy psycopg2-binary
# MySQL async
uv add sqlalchemy aiomysql
2. 使用非同步引擎設定資料庫
# 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. 模型定義
# 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. 非同步查詢
# 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. 與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. 生命週期事件-資料庫初始化
# 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)
總結
本課學到：
- 非同步引擎：使用asyncpg連接資料庫，連接池
- 聲明式模型：新的Mapped[]和mapped_column()樣式
- 人際關係：一對一、一對多、多對多
- 非同步查詢：select()、where()、join()、預先載入
- 儲存庫模式：標準CRUD操作
- FastAPI集成：資料庫會話的 Depends()
下一篇文章將指導 Alembic 遷移和資料庫播種。