SQLAlchemy 2: موتور، نشست و مدل‌های تایپ‌دار

۱. چرا پایگاه داده رابطه‌ای

داده پروژه تا حالا در دیکشنری بود و با هر راه‌اندازی دوباره سرور پاک می‌شد. یک پایگاه داده رابطه‌ای داده را در جدول‌ها نگه می‌دارد: هر ردیف یک رکورد، هر ستون یک فیلد با نوع مشخص. کلید اصلی هر ردیف را یکتا می‌کند و کلید خارجی یک ردیف را به ردیف جدول دیگری وصل می‌کند. مهم‌تر از همه، پایگاه داده قواعد را اجرا می‌کند: ایمیل تکراری، قیمت منفی یا مدرسی که وجود ندارد را حتی اگر کد ما اشتباه کند نمی‌پذیرد.

SQLAlchemy پرکاربردترین کتابخانه پایتون برای کار با این پایگاه‌هاست. در این دوره فقط نسخه ۲ و سبک تایپ‌دار آن را یاد می‌گیریم.

۲. موتور و نشست

from sqlalchemy import create_engine, text
from sqlalchemy.orm import Session

engine = create_engine('sqlite:///training.db', echo=True)


def main() -> None:
    # `with` = context manager: the session is closed even if an error happens.
    with Session(engine) as session:
        version = session.execute(text('select sqlite_version()')).scalar_one()
        print('SQLite version:', version)

        session.execute(
            text('create table if not exists notes (id integer primary key, body text)')
        )
        session.execute(text('insert into notes (body) values (:body)'), {'body': 'hello'})
        session.commit()

        rows = session.execute(text('select id, body from notes')).all()
        print(rows)


if __name__ == '__main__':
    main()

Engine و Session: نمونه 01_engine_and_session/app.py

• Engine یعنی «چطور به پایگاه داده برسیم». یک بار در کل برنامه ساخته می‌شود و مجموعه‌ای از اتصال‌ها را نگه می‌دارد. نشانی می‌گوید کدام پایگاه داده و کدام درایور: sqlite:///training.db یا postgresql+psycopg://user:pass@localhost/db.
• Session یعنی «یک گفت‌وگو با پایگاه داده». کوتاه‌عمر است: باز کن، کار کن، commit کن، ببند. در FastAPI برای هر درخواست یکی.
• echo=True هر دستور SQL را چاپ می‌کند. در زمان یادگیری روشنش بگذارید تا همیشه بدانید واقعاً چه اتفاقی می‌افتد.

۳. مدل‌ها: کلاس‌های پایتون که جدول را توصیف می‌کنند

class Base(DeclarativeBase):
    pass


def utc_now() -> datetime:
    return datetime.now(UTC)


class User(Base):
    __tablename__ = 'users'

    id: Mapped[int] = mapped_column(primary_key=True)
    email: Mapped[str] = mapped_column(String(255), unique=True, index=True)
    full_name: Mapped[str] = mapped_column(String(120))
    role: Mapped[str] = mapped_column(String(20), default='student')
    bio: Mapped[str | None] = mapped_column(Text)  # nullable, because of `| None`
    created_at: Mapped[datetime] = mapped_column(default=utc_now)


class Course(Base):
    __tablename__ = 'courses'

    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(200))
    price: Mapped[int]  # no mapped_column needed: the hint says everything
    capacity: Mapped[int]
    instructor_id: Mapped[int] = mapped_column(ForeignKey('users.id'))

DeclarativeBase، Mapped و mapped_column: نمونه 02_declarative_models/app.py

در سبک تایپ‌دار، راهنمای نوع خودِ تعریف ستون است:

آموزش‌های قدیمی می‌نویسند email = Column(String, unique=True). هنوز کار می‌کند، اما ویرایشگر و mypy نمی‌دانند user.email یک str است. با Mapped[str] می‌دانند: تکمیل خودکار و بررسی نوع، رایگان.

۴. درج و پرس‌وجو

from sqlalchemy import String, create_engine, func, select
from sqlalchemy.orm import DeclarativeBase, Mapped, Session, mapped_column


class Base(DeclarativeBase):
    pass


class Course(Base):
    __tablename__ = 'courses'

    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(200))
    price: Mapped[int]
    level: Mapped[str] = mapped_column(String(20), default='beginner')

    def __repr__(self) -> str:
        return f'Course({self.id}, {self.title!r}, {self.price:,})'


engine = create_engine('sqlite:///training.db')


def main() -> None:
    Base.metadata.drop_all(engine)
    Base.metadata.create_all(engine)

    with Session(engine) as session:
        python = Course(title='Python Basics', price=2_500_000)
        session.add(python)
        session.add_all(
            [
                Course(title='FastAPI', price=4_800_000, level='advanced'),
                Course(title='HTML and CSS', price=1_900_000),
            ]
        )
        print('before commit, id =', python.id)
        session.commit()
        print('after commit,  id =', python.id)

    with Session(engine) as session:
        print(session.get(Course, 2))
        print(session.get(Course, 99))

        cheap = session.scalars(
            select(Course).where(Course.price < 3_000_000).order_by(Course.price)
        )
        print(cheap.all())

        count = session.scalar(select(func.count()).select_from(Course))
        print('courses:', count)

        titles = session.scalars(select(Course.title).where(Course.level == 'beginner')).all()
        print(titles)


if __name__ == '__main__':
    main()

add، commit، get و select: نمونه 03_insert_and_query/app.py

before commit, id = None
after commit,  id = 1
Course(2, 'FastAPI', 4,800,000)
None
[Course(3, 'HTML and CSS', 1,900,000), Course(1, 'Python Basics', 2,500,000)]
courses: 3

select(Course) فقط یک شیء پرس‌وجو می‌سازد و تا وقتی نشست اجرایش نکند هیچ چیزی اجرا نمی‌شود. بعد از commit شیءهای تازه id دارند.

۵. قیدها و پیش‌فرض‌ها: پایگاه داده نه می‌گوید

Pydantic درخواست را بررسی می‌کند (فصل ۳). پایگاه داده آخرین خط دفاع است: داده را حتی در برابر باگ‌های خود ما و برنامه‌های دیگر حفظ می‌کند.

class User(Base):
    __tablename__ = 'users'

    id: Mapped[int] = mapped_column(primary_key=True)
    email: Mapped[str] = mapped_column(String(255), unique=True)
    is_active: Mapped[bool] = mapped_column(default=True)
    created_at: Mapped[datetime] = mapped_column(default=utc_now, server_default=func.now())
    updated_at: Mapped[datetime] = mapped_column(default=utc_now, onupdate=utc_now)


class Course(Base):
    __tablename__ = 'courses'
    __table_args__ = (
        CheckConstraint('price >= 0', name='price_not_negative'),
        CheckConstraint('capacity > 0', name='capacity_positive'),
    )

    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(200))
    price: Mapped[int]
    capacity: Mapped[int]
    instructor_id: Mapped[int] = mapped_column(ForeignKey('users.id'))


engine = create_engine('sqlite:///training.db')


@event.listens_for(engine, 'connect')
def enable_foreign_keys(connection, _record) -> None:  # type: ignore[no-untyped-def]
    """SQLite ignores foreign keys unless you ask. PostgreSQL always checks them."""
    connection.execute('PRAGMA foreign_keys=ON')


def try_to_save(session: Session, obj: object) -> None:
    session.add(obj)
    try:
        session.commit()
        print('saved   ', obj.__class__.__name__)
    except IntegrityError as error:
        session.rollback()  # REQUIRED after a failed commit
        print('REFUSED ', error.orig)

unique، CheckConstraint، ForeignKey و IntegrityError: نمونه 04_constraints_and_defaults/app.py

saved    User
REFUSED  UNIQUE constraint failed: users.email
REFUSED  CHECK constraint failed: price_not_negative
REFUSED  FOREIGN KEY constraint failed
saved    Course

۶. PostgreSQL روی کامپیوتر خودتان: فقط نشانی عوض می‌شود

SQLite یک فایل است و به چیزی نیاز ندارد. PostgreSQL یک سرور پایگاه داده است: به شکل یک سرویس در پس‌زمینه اجرا می‌شود و برنامه‌ها از راه شبکه به آن وصل می‌شوند، اینجا به localhost و درگاه 5432. آن را یک بار نصب می‌کنید و از آن به بعد هر جلسه فقط با عوض کردن DATABASE_URL از آن استفاده می‌کند.

نصب روی ویندوز ۱۱

در PowerShell این فرمان را بزنید، یا نصب‌کننده را از postgresql.org بگیرید:

winget install --id PostgreSQL.PostgreSQL.17

نصب‌کننده یک رمز برای کاربر مدیر یعنی **postgres می‌خواهد؛ آن را یادداشت کنید. درگاه را همان 5432** بگذارید و اجزای *PostgreSQL Server*، *pgAdmin 4* و *Command Line Tools* را نگه دارید. PostgreSQL از این به بعد یک سرویس ویندوز به نام postgresql-x64-17 است که با ویندوز روشن می‌شود. برای استفاده از psql در ترمینال، یک بار پوشه آن را به PATH اضافه کنید و ترمینال تازه‌ای باز کنید:

setx PATH "$env:PATH;C:\Program Files\PostgreSQL\17\bin"

نصب روی اوبونتو

sudo apt update
sudo apt install postgresql
sudo systemctl enable --now postgresql    # start now and at every boot
systemctl status postgresql               # should say "active"

اوبونتو کاربر مدیر postgres را بدون رمز می‌سازد و با sudo -u postgres از آن استفاده می‌کنید.

ساختن کاربر و پایگاه داده دوره

-- Create the user and the database of the course (run once, as the "postgres" superuser).
--   Windows:  psql -U postgres -f setup.sql
--   Ubuntu:   sudo -u postgres psql < setup.sql
CREATE ROLE training WITH LOGIN PASSWORD 'training';
CREATE DATABASE training OWNER training;

یک کاربر و یک پایگاه داده: 05_postgresql/setup.sql

psql -U postgres -f setup.sql             # Windows (asks the postgres password)
sudo -u postgres psql < setup.sql         # Ubuntu
python app.py
postgresql -> [(1, 'FastAPI')]

نشانی اتصال دوره از این به بعد این است: postgresql+psycopg://training:training@localhost:5432/training

درون psql، فرمان \dt جدول‌ها را فهرست می‌کند، \d courses یک جدول را توصیف می‌کند و \q خارج می‌شود.

خطاهای رایج اتصال و معنای آنها:

۷. FastAPI و پایگاه داده: یک نشست برای هر درخواست

وابستگی yield فصل چهارم حالا یک نشست واقعی باز می‌کند:

def get_db() -> Iterator[Session]:
    with SessionLocal() as session:
        yield session


DbSession = Annotated[Session, Depends(get_db)]
Base.metadata.create_all(engine)
app = FastAPI()


@app.post('/courses', status_code=status.HTTP_201_CREATED)
def create_course(data: CourseCreate, db: DbSession) -> CourseRead:
    course = Course(**data.model_dump())
    db.add(course)
    db.commit()
    db.refresh(course)  # reload generated values (the id) from the database
    return CourseRead.model_validate(course)


@app.get('/courses')
def list_courses(db: DbSession) -> list[CourseRead]:
    courses = db.scalars(select(Course).order_by(Course.id)).all()
    return [CourseRead.model_validate(course) for course in courses]


@app.get('/courses/{course_id}')
def get_course(course_id: int, db: DbSession) -> CourseRead:
    course = db.get(Course, course_id)
    if course is None:
        raise HTTPException(status.HTTP_404_NOT_FOUND, detail='Course not found')
    return CourseRead.model_validate(course)

get_db و endpointها: نمونه 06_fastapi_with_database/app.py

endpoint یک شیء ORM از پایگاه داده می‌گیرد و آن را از راه یک طرح‌واره Pydantic با from_attributes=True برمی‌گرداند. مدل ORM می‌گوید داده چگونه ذخیره می‌شود؛ طرح‌واره می‌گوید کلاینت چه می‌بیند. db.refresh(course) مقدارهای تولیدشده در پایگاه داده، مثل id، را دوباره می‌خواند.

۸. پروژه: API آموزشگاه روی پایگاه داده

store.py حذف می‌شود و دو فایل تازه جایش را می‌گیرند:

from collections.abc import Iterator
from datetime import datetime

from sqlalchemy import DateTime, create_engine, event
from sqlalchemy.orm import DeclarativeBase, Session, sessionmaker

from app.config import get_settings

settings = get_settings()
engine = create_engine(settings.database_url, echo=settings.database_echo)
SessionLocal = sessionmaker(engine)

if engine.dialect.name == 'sqlite':

    @event.listens_for(engine, 'connect')
    def enable_sqlite_foreign_keys(connection, _record) -> None:  # type: ignore[no-untyped-def]
        connection.execute('PRAGMA foreign_keys=ON')


class Base(DeclarativeBase):
    # Every Mapped[datetime] column keeps the time zone (TIMESTAMP WITH TIME ZONE).
    # Without it PostgreSQL stores "naive" times and asyncpg refuses aware ones.
    type_annotation_map = {datetime: DateTime(timezone=True)}


def get_db() -> Iterator[Session]:
    """One session per request; closed (and rolled back if needed) afterwards."""
    with SessionLocal() as session:
        yield session

موتور، کارخانه نشست و Base: 07_project/app/database.py

class User(Base):
    __tablename__ = 'users'

    id: Mapped[int] = mapped_column(primary_key=True)
    email: Mapped[str] = mapped_column(String(255), unique=True, index=True)
    full_name: Mapped[str] = mapped_column(String(120))
    role: Mapped[str] = mapped_column(String(20), default='student')
    is_active: Mapped[bool] = mapped_column(default=True)
    created_at: Mapped[datetime] = mapped_column(default=utc_now, server_default=func.now())
    updated_at: Mapped[datetime] = mapped_column(default=utc_now, onupdate=utc_now)


class Course(Base):
    __tablename__ = 'courses'
    __table_args__ = (
        CheckConstraint('price >= 0', name='price_not_negative'),
        CheckConstraint('capacity > 0', name='capacity_positive'),
    )

    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(200), index=True)
    description: Mapped[str] = mapped_column(Text, default='')
    price: Mapped[int]
    capacity: Mapped[int]
    level: Mapped[str] = mapped_column(String(20), default='beginner')
    is_active: Mapped[bool] = mapped_column(default=True)
    instructor_id: Mapped[int] = mapped_column(ForeignKey('users.id'), index=True)
    created_at: Mapped[datetime] = mapped_column(default=utc_now, server_default=func.now())
    updated_at: Mapped[datetime] = mapped_column(default=utc_now, onupdate=utc_now)


class Enrollment(Base):
    __tablename__ = 'enrollments'
    __table_args__ = (
        UniqueConstraint('student_id', 'course_id', name='one_enrollment_per_course'),
    )

    id: Mapped[int] = mapped_column(primary_key=True)
    student_id: Mapped[int] = mapped_column(ForeignKey('users.id', ondelete='CASCADE'))
    course_id: Mapped[int] = mapped_column(ForeignKey('courses.id', ondelete='CASCADE'))
    status: Mapped[str] = mapped_column(String(20), default='active')
    created_at: Mapped[datetime] = mapped_column(default=utc_now, server_default=func.now())

کاربر، دوره و ثبت‌نام: 07_project/app/models.py

سرویس‌ها حالا به جای دیکشنری یک Session می‌گیرند. امضای متدها، آرگومان‌ها و استثناها همان فصل چهارم است؛ فقط درونشان عوض شده و روترها هیچ تغییری نکرده‌اند.

    def create(self, data: CourseCreate) -> Course:
        instructor = self.db.get(User, data.instructor_id)
        if instructor is None or instructor.role != 'instructor':
            raise BusinessRuleError('instructor_id must be an instructor')
        course = Course(**data.model_dump())
        self.db.add(course)
        self.db.commit()
        self.db.refresh(course)
        return course

ساختن دوره با نشست: 07_project/app/services.py

در این جلسه ساختن و خواندن کار می‌کند. ویرایش و حذف با SQLAlchemy موضوع فصل ششم است.

۹. خلاصه فصل