اگر تا امروز با sqlite3 یا اجرای مستقیم کوئری‌های SQL کار کرده‌ای و با هر تغییر در مدل مجبور شده‌ای دیتابیس را دستی حذف و از نو بسازی، این آموزش دقیقاً نقطه پایان آن دردسر است. SQLAlchemy 2.0 با سینتکس مدرن و تایپ‌محور خود و Alembic به‌عنوان ابزار رسمی مهاجرت نسخه‌ای، به تو اجازه می‌دهند مدل‌های پایتونی را به جدول تبدیل کنی، ارتباط‌ها را شفاف تعریف کنی و هر تغییر اسکیما را با یک revision ایمن به تولید منتقل کنی — بدون از دست رفتن داده.

SQLAlchemy 2.0 و Alembic چیست و چرا بهترین انتخاب برای دیتابیس پایتون است؟

SQLAlchemy 2.0 یک ORM و لایه‌ی Core مدرن برای کار با دیتابیس در پایتون است که با DeclarativeBase، تایپ‌هینت و API یکپارچه‌ی 2.0 تجربه‌ای شبیه نوشتن کد پایتونی تمیز می‌دهد؛ Alembic هم مهاجرت‌های نسخه‌ای خودکار می‌سازد و با دستورات revision و upgrade تغییرات را بدون حذف داده اعمال می‌کند. در ادامه نصب، مدل‌سازی، Session مدرن و مهاجرت را گام‌به‌گام اجرا می‌کنیم.

فهرست مطالب

پیش‌نیازها و نصب SQLAlchemy 2.0 و Alembic
اولین اتصال و ساخت Base مدرن در SQLAlchemy 2.0
تعریف مدل‌ها و کار با Session به سبک 2.0
مهاجرت خودکار با Alembic؛ از init تا upgrade
رابطه‌ها، کوئری و تراکنش امن
مقایسه SQLAlchemy با روش‌های دیگر کار با دیتابیس
اشتباهات رایج هنگام کار با SQLAlchemy و Alembic
سوالات متداول
جمع‌بندی

پیش‌نیازها و نصب SQLAlchemy 2.0 و Alembic

برای این آموزش به پایتون 3.10 به بالا، یک محیط مجازی تمیز و یک دیتابیس نیاز داری. برای شروع SQLite کافی است؛ بعداً بدون تغییر کد می‌توانی به PostgreSQL یا MySQL سوئیچ کنی — این یکی از بزرگ‌ترین مزیت‌های ORM است.

نصب با uv و pip

اگر از uv استفاده می‌کنی که این روزها استاندارد سریع پایتون شده، نصب در چند ثانیه انجام می‌شود:

uv venv
source .venv/bin/activate  # ویندوز: .venv\Scripts\activate
uv pip install "SQLAlchemy>=2.0" alembic psycopg[binary]
# برای SQLite نیازی به psycopg نیست

با pip کلاسیک هم همین است:

pip install "SQLAlchemy>=2.0" alembic

نسخه را چک کن تا مطمئن شوی روی شاخه‌ی 2.0 هستی:

python -c "import sqlalchemy; print(sqlalchemy.__version__)"

ساختار پروژه پیشنهادی

یک ساختار مینیمال اما حرفه‌ای که هم برای اسکریپت و هم برای FastAPI جواب می‌دهد:

app/
  __init__.py
  db.py          # engine و Session
  models.py      # مدل‌ها
  config.py
alembic/
alembic.ini
main.py

این جداسازی باعث می‌شود Alembic دقیقاً بداند مدل‌ها کجا هستند و هنگام autogenerate چیزی را جا نیندازد.

اولین اتصال و ساخت Base مدرن در SQLAlchemy 2.0

در SQLAlchemy 2.0 دیگر خبری از declarative_base() قدیمی نیست؛ حالا از DeclarativeBase ارث‌بری می‌کنیم و تایپ‌ها را صریح می‌نویسیم. این تغییر فقط سلیقه‌ای نیست؛ به mypy، تکمیل خودکار IDE و تولید مهاجرت دقیق کمک می‌کند.

# app/db.py
from sqlalchemy import create_engine
from sqlalchemy.orm import DeclarativeBase, sessionmaker

# برای شروع SQLite، برای تولید PostgreSQL بگذار
DATABASE_URL = "sqlite+pysqlite:///app.db"
# مثال PostgreSQL: "postgresql+psycopg://user:pass@localhost:5432/mydb"

engine = create_engine(DATABASE_URL, echo=True, future=True)
SessionLocal = sessionmaker(bind=engine, autoflush=False, autocommit=False, future=True)

class Base(DeclarativeBase):
    pass

نکات کلیدی:

echo=True فقط برای توسعه است تا SQL تولیدی را ببینی؛ در تولید خاموش کن.
future=True تضمین می‌کند API کاملاً 2.0 باشد.
برای کار ناهمگام می‌توانی بعداً به create_async_engine و async_sessionmaker مهاجرت کنی؛ اما منطق مدل‌ها بدون تغییر می‌ماند.

اگر تازه از کار با درخواست‌های HTTP ناهمگام آمده‌ای و می‌خواهی همین الگوی مدرن را در لایه‌ی شبکه هم تجربه کنی، آموزش HTTPX؛ جایگزین مدرن requests با Async مکمل خوبی برای این بخش است چون همان فلسفه‌ی async-first را دنبال می‌کند.

تعریف مدل‌ها و کار با Session به سبک 2.0

حالا دو مدل واقعی می‌سازیم: کاربر و وظیفه. از Mapped و mapped_column استفاده می‌کنیم تا هم تایپ‌چک درست کار کند و هم Alembic تغییرات را بهتر تشخیص دهد.

# app/models.py
from datetime import datetime
from sqlalchemy import String, ForeignKey, Text, func
from sqlalchemy.orm import Mapped, mapped_column, relationship
from app.db import Base

class User(Base):
    __tablename__ = "users"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(80))
    email: Mapped[str] = mapped_column(String(120), unique=True, index=True)
    created_at: Mapped[datetime] = mapped_column(server_default=func.now())
    tasks: Mapped[list["Task"]] = relationship(back_populates="owner", cascade="all, delete-orphan")

class Task(Base):
    __tablename__ = "tasks"
    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(150))
    description: Mapped[str | None] = mapped_column(Text, nullable=True)
    is_done: Mapped[bool] = mapped_column(default=False)
    owner_id: Mapped[int] = mapped_column(ForeignKey("users.id", ondelete="CASCADE"))
    owner: Mapped["User"] = relationship(back_populates="tasks")

ایجاد جدول‌ها و اولین تراکنش

برای تست سریع می‌توانی جدول‌ها را مستقیم بسازی (در پروژه واقعی این کار را به Alembic می‌سپاریم):

# main.py
from app.db import Base, engine, SessionLocal
from app.models import User, Task

Base.metadata.create_all(bind=engine)

with SessionLocal() as session:
    with session.begin():
        user = User(name="سارا", email="sara@example.com")
        session.add(user)
        session.flush()  # تا id بگیریم بدون commit
        session.add_all([
            Task(title="نوشتن مقاله SQLAlchemy", owner_id=user.id),
            Task(title="بررسی مهاجرت Alembic", owner_id=user.id, is_done=True),
        ])
    print("commit شد")

# خواندن به سبک 2.0 با select
from sqlalchemy import select

with SessionLocal() as session:
    stmt = select(Task).where(Task.is_done == False).join(User).where(User.email == "sara@example.com")
    tasks = session.scalars(stmt).all()
    for t in tasks:
        print(t.title, "->", t.owner.name)

تفاوت مهم 2.0 اینجاست: به‌جای session.query(Task).filter(...) از select(Task).where(...) و session.scalars() استفاده می‌کنیم. خواناتر، تایپ‌ایمن‌تر و سازگار با async است.

اگر با دیتاهای جدولی بزرگ کار می‌کنی و می‌خواهی همین داده‌ها را بعداً تحلیل کنی، تجربه‌ی آموزش Pandas؛ تحلیل داده‌ها از صفر تا حرفه‌ای به تو کمک می‌کند خروجی همین کوئری‌ها را مستقیم وارد DataFrame کنی.

مهاجرت خودکار با Alembic؛ از init تا upgrade

ساخت جدول با create_all برای نمونه اولیه خوب است اما در پروژه واقعی هر تغییر ستون باید نسخه‌گذاری شود. Alembic دقیقاً همین کار را می‌کند و به‌عنوان ابزار رسمی SQLAlchemy، تغییرات مدل را به اسکریپت مهاجرت تبدیل می‌کند.

گام 1: مقداردهی اولیه

alembic init alembic

سپس alembic.ini را باز کن و sqlalchemy.url را برابر همان DATABASE_URL بگذار. مهم‌تر از آن، alembic/env.py را ویرایش کن:

# alembic/env.py
from app.db import Base, DATABASE_URL
from app.models import User, Task  # حتما ایمپورت تا metadata پر شود
from sqlalchemy import pool
from sqlalchemy import create_engine

target_metadata = Base.metadata
config.set_main_option("sqlalchemy.url", DATABASE_URL)

گام 2: ساخت revision خودکار

alembic revision --autogenerate -m "create users and tasks"
# خروجی: alembic/versions/xxxx_create_users_and_tasks.py

فایل ساخته‌شده را باز کن و حتماً چشمی مرور کن؛ autogenerate هوشمند است اما تغییر نام ستون یا تغییر نوع داده را گاهی به‌صورت حذف و اضافه نشان می‌دهد.

گام 3: اعمال مهاجرت

alembic upgrade head
# بازگشت در صورت نیاز
alembic downgrade -1

از این لحظه هر بار که مدلی را تغییر دادی (مثلاً افزودن priority: Mapped[int] به Task)، فقط یک revision جدید بساز و upgrade کن. دیگر نیازی به حذف دیتابیس نیست و داده‌های قبلی سالم می‌ماند.

نکته طلایی برای تیم

فایل‌های alembic/versions/*.py را کامیت کن؛ تاریخچه‌ی دیتابیس بخشی از کد است.
برای CI، دستور alembic upgrade head را قبل از اجرای تست‌ها اجرا کن تا دیتابیس تست هم‌سان باشد.
اگر از SQLite به PostgreSQL مهاجرت می‌کنی، تفاوت نوع‌ها مثل String بدون طول را در Postgres جدی بگیر.

رابطه‌ها، کوئری و تراکنش امن

Relationship یک‌به‌چند با lazy بهینه

تعریف relationship در مثال بالا به‌صورت پیش‌فرض select است. برای جلوگیری از N+1، هنگام خواندن لیست تسک‌ها همراه با مالک، از selectinload استفاده کن:

from sqlalchemy.orm import selectinload
from sqlalchemy import select
from app.models import Task

stmt = select(Task).options(selectinload(Task.owner)).where(Task.is_done == False)
tasks = session.scalars(stmt).all()

تراکنش امن با session.begin()

همیشه تراکنش را با کانتکست منیجر ببند تا در صورت خطا rollback خودکار شود:

from sqlalchemy.exc import IntegrityError

try:
    with SessionLocal() as session:
        with session.begin():
            session.add(User(name="تکراری", email="sara@example.com"))  # ایمیل تکراری
except IntegrityError:
    print("ایمیل تکراری است، تراکنش rollback شد")

هرگز commit را دستی در چند جا پخش نکن؛ یک begin() برای هر واحد کاری کافی و امن است. برای عملیات حساس مالی، ایزولیشن لول را در create_engine تنظیم کن.

مقایسه SQLAlchemy با روش‌های دیگر کار با دیتابیس

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

روش	مزایا	محدودیت‌ها	بهترین کاربرد
sqlite3 / psycopg مستقیم	سبک، بدون وابستگی، کنترل کامل SQL	مدیریت دستی migration، تکرار کد، ریسک SQL Injection در صورت الحاق رشته	اسکریپت‌های یک‌بارمصرف و آموزشی
SQLAlchemy Core	ساخت SQL با پایتون، تایپ‌ایمن، قابل حمل بین دیتابیس‌ها	بدون نگاشت شی‌گرا، رابطه‌ها دستی	گزارش‌گیری و کوئری‌های پیچیده تحلیلی
SQLAlchemy ORM 2.0	مدل شی‌گرا، رابطه‌ها، Session واحد، سازگار با Alembic و async	منحنی یادگیری بیشتر، نیاز به درک Session	وب‌اپ، API و پروژه‌های بلندمدت
SQLModel / Pydantic + SQLAlchemy	اعتبارسنجی خودکار با Pydantic، یک مدل برای API و DB	وابستگی به اکوسیستم FastAPI، انعطاف کمتر در mapping پیشرفته	FastAPI که همزمان اعتبارسنجی API می‌خواهد

اگر پروژه‌ات از یک اسکریپت کوچک فراتر می‌رود و قرار است ماه‌ها نگهداری شود، ORM 2.0 + Alembic کم‌هزینه‌ترین انتخاب در بلندمدت است.

اشتباهات رایج هنگام کار با SQLAlchemy و Alembic

این خطاها را بارها در پروژه‌های واقعی دیده‌ایم؛ با دانستن‌شان از ابتدا جلوی باگ‌های پرهزینه را می‌گیری:

فراموشی ایمپورت مدل‌ها در env.py: اگر User و Task را در env.py ایمپورت نکنی، target_metadata خالی می‌ماند و autogenerate مهاجرت خالی می‌سازد. همیشه همه‌ی مدل‌ها را ایمپورت کن.
ترکیب create_all و Alembic: بعد از راه‌اندازی Alembic دیگر Base.metadata.create_all را در کد تولید صدا نزن؛ فقط alembic upgrade head را اجرا کن تا تاریخچه ناهمگام نشود.
استفاده از API قدیمی 1.x: نوشتن session.query(...) در کد 2.0 هم کار می‌کند اما deprecation warning می‌دهد و با async ناسازگار است. به select() مهاجرت کن.
عدم بررسی فایل revision تولیدشده: اعتماد کور به autogenerate گاهی ستون را drop/add می‌کند و داده را از بین می‌برد. همیشه upgrade و downgrade را در دیتابیس تست امتحان کن.
نگه‌داشتن Session باز: Session را به‌صورت گلوبال نگه ندار. با with SessionLocal() as session و session.begin() چرخه‌ی عمر را کوتاه و قابل پیش‌بینی کن.

سوالات متداول

تفاوت SQLAlchemy 2.0 با نسخه 1.4 چیست و آیا باید مهاجرت کنم؟

نسخه 2.0 API را یکپارچه و تایپ‌محور کرده؛ select() جای query() را گرفته، DeclarativeBase و Mapped[] اجباری شده و پشتیبانی از async رسمی است. اگر پروژه‌ات فعال است، مهاجرت تدریجی توصیه می‌شود چون 1.4 دیگر به‌روزرسانی ویژگی نمی‌گیرد و مستندات جدید فقط 2.0 را پوشش می‌دهد.

آیا Alembic بدون SQLAlchemy هم کار می‌کند؟

بله، می‌توانی Alembic را با SQL خام هم استفاده کنی، اما بیشترین ارزش آن زمانی است که target_metadata از SQLAlchemy پر شود تا --autogenerate تغییرات مدل را تشخیص دهد. بدون ORM، باید مهاجرت‌ها را دستی بنویسی.

برای پروژه کوچک هم Alembic لازم است؟

اگر پروژه قرار است فقط یک هفته زنده باشد، نه. اما به‌محض اینکه حتی یک تغییر ستون در آینده محتمل است، هزینه‌ی راه‌اندازی Alembic (حدود 5 دقیقه) بسیار کمتر از بازسازی دستی دیتابیس و از دست رفتن داده در تولید است.

چطور از SQLite در توسعه به PostgreSQL در تولید سوئیچ کنم؟

کافی است DATABASE_URL را عوض کنی؛ مدل‌ها بدون تغییر می‌مانند. فقط مراقب تفاوت‌ها باش: طول String، نوع JSON، و server_defaultها در Postgres سخت‌گیرتر است. یک بار alembic upgrade head را روی Postgres تست کن.

چگونه مهاجرت اشتباه را به‌صورت ایمن برگردانم؟

اگر revision هنوز به تولید نرفته، alembic downgrade -1 و سپس حذف فایل revision کافی است. اگر به تولید رفته، یک revision جدید بساز که تغییر را معکوس کند؛ هرگز تاریخچه‌ی منتشرشده را بازنویسی نکن.

آیا SQLAlchemy برای کار ناهمگام مناسب است؟

بله؛ SQLAlchemy 2.0 دارای create_async_engine و AsyncSession است و با FastAPI async عالی کار می‌کند. فقط درایور ناهمگام مثل asyncpg برای Postgres لازم داری و باید از await session.scalars() استفاده کنی.

جمع‌بندی

حالا یک نقشه‌ی راه شفاف داری:

محیط مجازی و SQLAlchemy 2.0 + Alembic را نصب کن و نسخه را چک کن.
Base مدرن با DeclarativeBase و Mapped[] بساز و engine را متمرکز در db.py نگه دار.
مدل‌های User و Task را تعریف کن و با Session و select() تراکنش امن بنویس.
Alembic را init کن، env.py را به Base.metadata وصل کن و اولین revision --autogenerate و upgrade head را اجرا کن.
رابطه‌ها را با selectinload بهینه کن و هر تغییر بعدی را فقط با revision جدید اعمال کن.
قبل از هر upgrade، فایل مهاجرت را مرور و در دیتابیس تست downgrade را امتحان کن.

با همین چرخه، دیتابیس‌ات نسخه‌پذیر، قابل تست و آماده‌ی رشد می‌شود — چه روی SQLite محلی و چه روی PostgreSQL تولید. از اینجا می‌توانی سراغ احراز هویت، صفحه‌بندی کوئری‌ها یا اتصال به FastAPI بروی بدون اینکه نگران اسکیمای فردا باشی.

آموزش SQLAlchemy 2.0 و Alembic؛ کار با دیتابیس در پایتون