CRUD واقعی با SQLAlchemy 2

۱. یک مدل مشترک برای تمرین

پوشه‌های 01_queries و 02_changes هر کدام یک models.py دارند با کاربر و دوره و تابع reset_database() که هر بار داده نمونه یکسانی می‌سازد. اسکریپت‌های شماره‌دار را به هر ترتیبی می‌توانید اجرا کنید.

۲. select و where

from sqlalchemy import or_, select
from sqlalchemy.orm import Session

from models import Course, engine, reset_database


def main() -> None:
    reset_database()
    with Session(engine) as session:
        stmt = select(Course).where(Course.price < 3_000_000).order_by(Course.price)
        print(stmt)  # the SQL, with placeholders
        print(session.scalars(stmt).all())

        both = select(Course).where(Course.level == 'advanced', Course.instructor_id == 1)
        print(session.scalars(both).all())

        either = select(Course).where(or_(Course.level == 'beginner', Course.price > 4_000_000))
        print(session.scalars(either).all())

        in_list = select(Course.title).where(Course.level.in_(['beginner', 'intermediate']))
        print(session.scalars(in_list).all())

        like = select(Course).where(Course.title.ilike('%python%'), Course.is_active)
        print(session.scalars(like).all())


if __name__ == '__main__':
    main()

ساختن پرس‌وجو و اجرای آن: 01_queries/01_select_and_where.py

Course.price < 3_000_000 عددها را مقایسه نمی‌کند؛ یک تکه SQL می‌سازد: courses.price < :price_1. هر پرس‌وجو را می‌توان چاپ کرد و SQL آن را خواند.

۳. scalar، scalars یا execute؟

from sqlalchemy import func, select
from sqlalchemy.exc import MultipleResultsFound, NoResultFound
from sqlalchemy.orm import Session

from models import Course, engine, reset_database


def main() -> None:
    reset_database()
    with Session(engine) as session:
        courses = session.scalars(select(Course).order_by(Course.id)).all()
        print('scalars   ->', courses[:2], '...')

        count = session.scalar(select(func.count()).select_from(Course))
        print('scalar    ->', count)

        rows = session.execute(select(Course.title, Course.price).where(Course.price > 3_000_000))
        for row in rows:
            print('execute   ->', row.title, row.price)

        print('get       ->', session.get(Course, 3), session.get(Course, 99))

        one = session.execute(select(Course).where(Course.title == 'FastAPI')).scalar_one()
        print('scalar_one->', one)

        for title in ['Nothing', None]:
            stmt = select(Course).where(Course.title == title) if title else select(Course)
            try:
                session.execute(stmt).scalar_one()
            except (NoResultFound, MultipleResultsFound) as error:
                print('error     ->', type(error).__name__)


if __name__ == '__main__':
    main()

سه روش گرفتن نتیجه: 01_queries/02_scalar_scalars_execute.py

۴. شمردن و گروه‌بندی در خود پایگاه داده

هرگز ده هزار ردیف را برای شمردن به پایتون نیاورید:

from sqlalchemy import func, select
from sqlalchemy.orm import Session

from models import Course, User, engine, reset_database


def main() -> None:
    reset_database()
    with Session(engine) as session:
        print('active courses:', session.scalar(select(func.count()).where(Course.is_active)))
        print('average price :', round(session.scalar(select(func.avg(Course.price))) or 0))
        print('cheapest      :', session.scalar(select(func.min(Course.price))))

        per_level = session.execute(
            select(Course.level, func.count().label('n')).group_by(Course.level).order_by('level')
        )
        for level, n in per_level:
            print(f'  {level:<13} {n}')

        per_teacher = session.execute(
            select(User.full_name, func.count(Course.id), func.sum(Course.price))
            .join(Course, Course.instructor_id == User.id)
            .group_by(User.id)
        )
        for name, n, total in per_teacher:
            print(f'  {name:<13} {n} courses, {total:,} IRT')


if __name__ == '__main__':
    main()

count، avg و group_by: 01_queries/03_aggregates_and_groups.py

۵. ساختن جست‌وجو از فیلترهای اختیاری

هر .where() یک پرس‌وجوی تازه برمی‌گرداند. پس تابع جست‌وجو فقط شرط‌هایی را اضافه می‌کند که کاربر خواسته است:

def search_courses(
    session: Session,
    *,
    q: str | None = None,
    level: str | None = None,
    min_price: int | None = None,
    max_price: int | None = None,
    only_active: bool = True,
) -> list[Course]:
    stmt = select(Course)
    if q:
        stmt = stmt.where(Course.title.ilike(f'%{q}%'))
    if level:
        stmt = stmt.where(Course.level == level)
    if min_price is not None:
        stmt = stmt.where(Course.price >= min_price)
    if max_price is not None:
        stmt = stmt.where(Course.price <= max_price)
    if only_active:
        stmt = stmt.where(Course.is_active)
    return list(session.scalars(stmt.order_by(Course.price)))

جست‌وجو با فیلترهای اختیاری: 01_queries/04_search_and_filter.py

۶. ویرایش و حذف

در شیوه ORM، شیء را بارگذاری می‌کنیم، صفت‌هایش را عوض می‌کنیم و commit می‌کنیم. نشست تغییرات را دنبال می‌کند و فقط ستون‌های تغییرکرده را می‌نویسد:

from sqlalchemy import select, update
from sqlalchemy.orm import Session

from models import Course, engine, reset_database


def main() -> None:
    reset_database()
    with Session(engine) as session:
        course = session.get(Course, 1)
        assert course is not None
        course.price = 2_700_000
        course.title = 'Python Basics (2026)'
        print('dirty before commit:', session.dirty)
        session.commit()
        print('after commit       :', session.get(Course, 1))

        # "apply these changes" from a dict -- exactly what a PATCH endpoint does
        changes = {'level': 'intermediate', 'price': 3_000_000}
        for field, value in changes.items():
            setattr(course, field, value)
        session.commit()
        print('patched            :', course, course.level)

        result = session.execute(
            update(Course).where(Course.level == 'advanced').values(price=Course.price * 110 // 100)
        )
        session.commit()
        print('bulk updated rows  :', result.rowcount)
        print(session.scalars(select(Course).where(Course.level == 'advanced')).all())


if __name__ == '__main__':
    main()

ویرایش شیء، setattr برای PATCH و ویرایش گروهی: 02_changes/01_update.py

def delete_course(session: Session, course_id: int) -> bool:
    course = session.get(Course, course_id)
    if course is None:
        return False  # the API turns this into 404
    session.delete(course)
    session.commit()
    return True

حذف امن: 02_changes/02_delete.py

حذف نرم یعنی به جای حذف واقعی، is_active = False. تاریخچه، مثل پرداخت‌ها و ثبت‌نام‌ها، درست می‌ماند.

۷. تراکنش: همه یا هیچ

همه چیز بین دو commit یک تراکنش است. اگر وسط کار چیزی شکست بخورد، rollback() همه را برمی‌گرداند:

def hand_over(old_id: int, new_id: int, fail_half_way: bool = False) -> None:
    with Session(engine) as session, session.begin():
        for course in session.scalars(select(Course).where(Course.instructor_id == old_id)):
            course.instructor_id = new_id
            if fail_half_way:
                raise RuntimeError('network failure after moving one course')
        old = session.get(User, old_id)
        assert old is not None
        session.delete(old)


def report() -> None:
    with Session(engine) as session:
        users = session.scalar(select(func.count()).select_from(User))
        ali = session.scalar(select(func.count()).where(Course.instructor_id == 1))
        print(f'  users={users}, courses of Ali={ali}')


def main() -> None:
    reset_database()
    report()
    try:
        hand_over(old_id=1, new_id=2, fail_half_way=True)
    except RuntimeError as error:
        print('failed:', error)
    report()  # nothing changed: the first course was moved back by the rollback

    hand_over(old_id=1, new_id=2)
    print('done')
    report()


if __name__ == '__main__':
    main()

مدرسی که می‌رود: 02_changes/03_transactions.py

  users=3, courses of Ali=3
failed: network failure after moving one course
  users=3, courses of Ali=3
done
  users=2, courses of Ali=0

۸. ایمیل تکراری: اول بررسی کن، باز هم IntegrityError را بگیر

class DuplicateEmailError(Exception):
    pass


def register(session: Session, email: str, full_name: str, check_first: bool = True) -> User:
    email = email.lower()
    if check_first and session.scalar(select(User).where(User.email == email)):
        raise DuplicateEmailError(f'{email} is already registered (checked in Python)')
    user = User(email=email, full_name=full_name)
    session.add(user)
    try:
        session.commit()
    except IntegrityError as error:
        session.rollback()
        raise DuplicateEmailError(f'{email} is already registered (database said no)') from error
    return user

دو لایه دفاع: 02_changes/04_integrity_errors.py

1. بررسی در سرویس («این ایمیل گرفته شده؟») پیام دوستانه ۴۰۹ می‌دهد.
2. قید پایگاه داده تور ایمنی است. دو درخواست می‌توانند هم‌زمان از بررسی عبور کنند؛ فقط پایگاه داده جلوی دومی را می‌گیرد. پس IntegrityError را هم بگیرید، rollback() کنید و باز ۴۰۹ بدهید.

۹. چرخه عمر نشست

فایل 05_session_lifecycle.py با echo روشن نشان می‌دهد SQL کی فرستاده می‌شود:

• نقشه هویت: در یک نشست، get(Course, 1) دو بار همان شیء پایتون را برمی‌گرداند و بار دوم هیچ SQLی اجرا نمی‌شود.
• **flush()** تغییرات را به پایگاه داده می‌فرستد بدون اینکه تراکنش تمام شود؛ برای گرفتن زودهنگام id مفید است.
• **commit()** یعنی flush و پایان تراکنش. شیءها «منقضی» می‌شوند و دسترسی بعدی دوباره از پایگاه داده می‌خواند.
• **refresh(obj)** یک شیء را همین حالا دوباره می‌خواند.

۱۰. یک CRUD «کارا» با شش مشکل جدی

03_refactor_me/app.py کار می‌کند، تا وقتی دو چیز خراب شود:

1. یک نشست سراسری برای همه درخواست‌ها. بعد از یک commit ناموفق، مثلاً ثبت ایمیل تکراری، نشست خراب می‌شود و همه درخواست‌های بعدی با PendingRollbackError شکست می‌خورند.
2. جست‌وجو ورودی کاربر را به متن SQL می‌چسباند: تزریق SQL. نشانی /courses/search?q=%' OR is_hidden=1 -- دوره مخفی را نشان می‌دهد.
3. برای یافتن یک دوره، همه دوره‌ها بارگذاری و در پایتون جست‌وجو می‌شوند.
4. دوره ناموجود ۲۰۰ با {"error": ...} و حذف {"ok": false} برمی‌گرداند.
5. ویرایش یک دیکشنری خام می‌گیرد: بدون اعتبارسنجی، کلاینت می‌تواند id را هم عوض کند.
6. طرح‌واره پاسخ ندارد: همه ستون‌ها به کلاینت فرستاده می‌شوند.

۱۱. پروژه: CRUD کامل آموزشگاه

    def _apply(self, course: Course, changes: dict[str, Any]) -> Course:
        if 'title' in changes and changes['title'] != course.title:
            self._check_title_is_free(changes['title'], course.instructor_id)
        if 'capacity' in changes:
            active = EnrollmentService(self.db).active_count(course.id)
            if changes['capacity'] < active:
                raise BusinessRuleError(f'capacity can not be lower than {active} active students')
        for field, value in changes.items():
            setattr(course, field, value)
        self.db.commit()
        self.db.refresh(course)
        return course

اعمال تغییرات با قاعده‌های کسب‌وکار: 04_project/app/services.py

• دوره‌ها: PUT (جایگزینی)، PATCH (تغییر بخشی)، DELETE و جست‌وجو با q، level، min_price، max_price و is_active.
• PATCH با {"title": null} یا بدنه خالی رد می‌شود.
• کاربران: PATCH و DELETE (برای مدرسی که هنوز دوره دارد ۴۰۹).
• ثبت‌نام: DELETE لغو می‌کند و ثبت‌نام دوباره همان ردیف را فعال می‌کند.
• ظرفیت نمی‌تواند از تعداد دانشجویان فعال کمتر شود.
• handler IntegrityError در exceptions.py آخرین تور ایمنی است.

۱۲. خلاصه فصل