Alembic، صفحه‌بندی، فیلتر و مرتب‌سازی

۱. چرا create_all کافی نیست

Base.metadata.create_all() فقط جدول‌هایی را می‌سازد که وجود ندارند. هرگز جدول موجود را تغییر نمی‌دهد و هیچ تاریخچه‌ای نگه نمی‌دارد. اگر امروز به مدل ستونی اضافه کنید، روی پایگاه داده‌ای که از قبل وجود دارد هیچ اتفاقی نمی‌افتد و برنامه با خطا مواجه می‌شود.

Alembic تاریخچه طرح پایگاه داده را به شکل فایل‌های پایتونی شماره‌دار نگه می‌دارد، همان‌طور که Git تاریخچه کد را نگه می‌دارد. هر نسخه یک upgrade() برای جلو رفتن و یک downgrade() برای برگشتن دارد. خود پایگاه داده در جدول کوچکی به نام alembic_version به خاطر می‌سپارد در کدام نسخه است.

۲. Alembic در پنج دقیقه

پوشه 01_alembic_basics دو نسخه واقعی دارد که با فرمان‌های زیر ساخته شده‌اند:

alembic init alembic
alembic revision --autogenerate -m "create courses" --rev-id 0001
alembic upgrade head
# add the `level` column to models.py, then:
alembic revision --autogenerate -m "add course level" --rev-id 0002
alembic upgrade head

"""add course level

Revision ID: 0002
Revises: 0001
Create Date: 2026-09-24 13:13:04.717301

"""

from collections.abc import Sequence

from alembic import op
import sqlalchemy as sa


# revision identifiers, used by Alembic.
revision: str = '0002'
down_revision: str | Sequence[str] | None = '0001'
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None


def upgrade() -> None:
    """Upgrade schema."""
    # ### commands auto generated by Alembic - please adjust! ###
    with op.batch_alter_table('courses', schema=None) as batch_op:
        batch_op.add_column(
            sa.Column('level', sa.String(length=20), server_default='beginner', nullable=False)
        )

    # ### end Alembic commands ###


def downgrade() -> None:
    """Downgrade schema."""
    # ### commands auto generated by Alembic - please adjust! ###
    with op.batch_alter_table('courses', schema=None) as batch_op:
        batch_op.drop_column('level')

    # ### end Alembic commands ###

نسخه‌ای که Alembic از روی تفاوت مدل و پایگاه داده ساخت: 01_alembic_basics/alembic/versions/0002_add_course_level.py

دو تغییر در env.py

پس از alembic init فقط دو جای alembic/env.py عوض می‌شود: target_metadata = Base.metadata تا autogenerate بتواند مدل‌ها را با پایگاه داده مقایسه کند، و config.set_main_option('sqlalchemy.url', ...) تا برنامه و مهاجرت‌ها از یک نشانی استفاده کنند. چون SQLite بیشتر دستورهای ALTER را پشتیبانی نمی‌کند، render_as_batch=True را هم اضافه می‌کنیم: Alembic جدول را در حالت «batch» کپی و دوباره می‌سازد.

۳. صفحه‌بندی

هرگز ده هزار ردیف را در یک پاسخ برنگردانید. پاسخ یک صفحه باید به کلاینت بگوید کجاست:

{"items": [...], "total": 45, "page": 3, "size": 10, "pages": 5}

@app.get('/courses')
def list_courses(
    db: Annotated[Session, Depends(get_db)],
    page: Annotated[int, Query(ge=1)] = 1,
    size: Annotated[int, Query(ge=1, le=100)] = 10,
) -> CoursePage:
    stmt = select(Course).order_by(Course.id)
    total = db.scalar(select(func.count()).select_from(Course)) or 0
    courses = db.scalars(stmt.offset((page - 1) * size).limit(size)).all()
    return CoursePage(
        items=[CourseRead.model_validate(c) for c in courses],
        total=total,
        page=page,
        size=size,
        pages=math.ceil(total / size),
    )

صفحه‌بندی offset و limit: 02_list_api/01_pagination.py

total به یک پرس‌وجوی دوم و ارزان نیاز دارد: count(*) با همان فیلترها. پرس‌وجوی صفحه‌بندی‌شده را همیشه بر اساس چیزی یکتا مرتب کنید، وگرنه یک ردیف ممکن است در دو صفحه بیاید یا در هیچ صفحه‌ای نیاید.

۴. جست‌وجو، فیلتر و مرتب‌سازی به شکل یک مدل

همه گزینه‌های فهرست در یک مدل Pydantic تعریف و با Annotated[CourseQuery, Query()] از رشته پرس‌وجو خوانده می‌شوند:

SortField = Literal['title', '-title', 'price', '-price', 'id', '-id']

SORT_COLUMNS = {
    'title': Course.title.asc(),
    '-title': Course.title.desc(),
    'price': Course.price.asc(),
    '-price': Course.price.desc(),
    'id': Course.id.asc(),
    '-id': Course.id.desc(),
}


class CourseQuery(BaseModel):
    model_config = ConfigDict(extra='forbid')

    q: str | None = Field(default=None, min_length=2, max_length=50)
    level: Literal['beginner', 'intermediate', 'advanced'] | None = None
    min_price: int | None = Field(default=None, ge=0)
    max_price: int | None = Field(default=None, ge=0)
    is_active: bool | None = True
    sort: SortField = 'id'
    page: int = Field(default=1, ge=1)
    size: int = Field(default=10, ge=1, le=100)

    @model_validator(mode='after')
    def check_prices(self) -> Self:
        if self.min_price is not None and self.max_price is not None:
            if self.min_price > self.max_price:
                raise ValueError('min_price is greater than max_price')
        return self

مدل پرس‌وجو و فهرست سفید مرتب‌سازی: 02_list_api/02_filter_and_sort.py

قواعد طراحی:

• همه پارامترها اختیاری، با پیش‌فرض معقول.
• همه چیز بررسی می‌شود: size=1000 یا حداقل قیمت بیشتر از حداکثر، ۴۲۲ می‌گیرد.
• مرتب‌سازی از فهرست سفید: کلاینت یک نام انتخاب می‌کند و ما آن را به یک ستون نگاشت می‌کنیم. متن کلاینت هرگز مستقیم به order_by نمی‌رود.
• extra='forbid': اشتباه تایپی مثل ?levle=advanced خطاست، نه چیزی که بی‌صدا نادیده گرفته شود.

۵. صفحه‌بندی را یک بار بنویسید

هر endpoint فهرستی همان چهار قدم را لازم دارد: شمردن، مرتب کردن، offset و limit، و بسته‌بندی در یک صفحه. جنریک‌های فصل اول اجازه می‌دهند یک تابع برای همه مدل‌ها کار کند:

class PageParams(BaseModel):
    page: int = Field(default=1, ge=1)
    size: int = Field(default=10, ge=1, le=100)


class Page[T](BaseModel):
    items: list[T]
    total: int
    page: int
    size: int

    @computed_field
    @property
    def pages(self) -> int:
        return math.ceil(self.total / self.size)


def paginate[S: BaseModel](
    db: Session,
    stmt: Select,
    params: PageParams,
    schema: type[S],  # type: ignore[type-arg]
) -> Page[S]:
    total = db.scalar(select(func.count()).select_from(stmt.order_by(None).subquery())) or 0
    rows = db.scalars(stmt.offset((params.page - 1) * params.size).limit(params.size)).all()
    return Page[S](
        items=[schema.model_validate(row) for row in rows],
        total=total,
        page=params.page,
        size=params.size,
    )

Page جنریک و paginate: 02_list_api/03_reusable_paginate.py

۶. پروژه: مهاجرت‌ها و یک API فهرستی واقعی

• create_all از app/main.py حذف شد. جدول‌ها را حالا مهاجرت‌ها می‌سازند: 0001_initial_tables کاربران، دوره‌ها و ثبت‌نام‌ها را می‌سازد و 0002_add_course_start_date یک ستون تازه start_date به دوره‌ها اضافه می‌کند؛ این همان تحول طرح در طول زمان است.
• GET /courses یک Page برمی‌گرداند با جست‌وجوی q، فیلترهای level، instructor_id، min_price، max_price، starts_after و is_active و مرتب‌سازی با sort=-price.
• GET /users هم با همان Page[T] و paginate() صفحه‌بندی می‌شود.

    def list_courses(self, query: CourseQuery) -> Page[CourseRead]:
        stmt = select(Course)
        if query.q:
            stmt = stmt.where(Course.title.ilike(f'%{query.q}%'))
        if query.level:
            stmt = stmt.where(Course.level == query.level)
        if query.instructor_id is not None:
            stmt = stmt.where(Course.instructor_id == query.instructor_id)
        if query.min_price is not None:
            stmt = stmt.where(Course.price >= query.min_price)
        if query.max_price is not None:
            stmt = stmt.where(Course.price <= query.max_price)
        if query.starts_after is not None:
            stmt = stmt.where(Course.start_date >= query.starts_after)
        if query.is_active is not None:
            stmt = stmt.where(Course.is_active == query.is_active)
        stmt = stmt.order_by(SORT_COLUMNS[query.sort], Course.id)  # id breaks ties
        return paginate(self.db, stmt, query, CourseRead)

ساختن پرس‌وجوی فهرست: 03_project/app/services.py

alembic upgrade head      # creates training.db from the migrations
python seed.py
python main.py

۷. خلاصه فصل