آموزش FastAPI و SQLAlchemy؛ ساخت API پایتون آماده Production
راهنمای پروژهمحور ساخت REST API با FastAPI، Pydantic و SQLAlchemy؛ از نصب و مدلسازی تا Session دیتابیس، مدیریت خطا، امنیت، تست و استقرار Production.
FastAPI یک Framework مدرن پایتون برای ساخت API است که از Type Hintهای استاندارد برای اعتبارسنجی ورودی، تولید Schema و مستندات OpenAPI استفاده میکند. اگر فقط یک Route نمونه بسازیم، شروع کار بسیار کوتاه است؛ اما API واقعی به ساختار پروژه، چرخه صحیح Session دیتابیس، مدلهای ورودی و خروجی جدا، مدیریت خطا، تست، امنیت و استقرار قابل تکرار نیاز دارد.
در این آموزش یک API مدیریت وظایف میسازیم که Create، Read و Update دارد و از FastAPI، Pydantic و SQLAlchemy 2 استفاده میکند. هدف کپیکردن چند خط کد نیست؛ هدف این است که بدانید هر لایه چه مسئولیتی دارد و نسخه Production چه تفاوتی با Demo دارد.
چرا FastAPI اکنون ارزش یادگیری دارد؟
در بررسی واقعی Google Trends ایران در ۹ سپتامبر ۲۰۲۶، Python میانگین علاقه ۷۴ در بازه ۹۰روزه و ۲۲ در بازه ۱۲ماهه داشت؛ یعنی تقاضای پایه قوی است. FastAPI در مقایسه مستقیم حجم کوچکتری داشت، اما Related Query عبارت fastapi python در ۹۰ روز ۳۰۰٪ رشد کرد، sqlalchemy به وضعیت Breakout رسید و fastapi documentation در ۱۲ ماه ۱۱۰٪ رشد داشت. عبارت python web framework نیز Breakout بود.
پیشنهادهای واقعی Google Search شامل fastapi tutorial، fastapi python، fastapi documentation، fastapi install و مقایسه با Flask و Django بود. پس این موضوع یک «زیرموضوع روبهرشد» در اکوسیستم پرتقاضای Python است، نه صرفاً یک واژه پرحجم عمومی.
FastAPI چه چیزی را خودکار میکند؟
- تبدیل Path، Query، Header و Body به Typeهای پایتون
- اعتبارسنجی داده با Pydantic و پاسخ خطای ساختاریافته
- تولید OpenAPI و رابط Swagger UI در مسیر
/docs - Dependency Injection برای دیتابیس، احراز هویت و سرویسها
- پشتیبانی طبیعی از تابعهای Sync و Async
- تعریف Response Model برای کنترل خروجی عمومی
این قابلیتها سرعت توسعه را بالا میبرند، ولی جای طراحی Domain، Transaction، Authorization و Monitoring را نمیگیرند. Type Hint قرارداد ورودی را روشن میکند؛ تصمیم اینکه چه کسی اجازه تغییر یک Task را دارد هنوز مسئولیت برنامه شماست.
FastAPI یا Flask یا Django؟
| معیار | FastAPI | Flask | Django |
|---|---|---|---|
| تمرکز اصلی | API مبتنی بر Type Hint | هسته سبک و انعطافپذیر | Framework کامل وب |
| اعتبارسنجی | Pydantic و Schema خودکار | با Extension یا کد دستی | Form و Serializer با ابزارهای مکمل |
| مستندات API | OpenAPI و Swagger خودکار | نیازمند افزونه | معمولاً با Django REST Framework |
| ORM داخلی | ندارد؛ انتخاب آزاد | ندارد؛ انتخاب آزاد | ORM یکپارچه |
| Async | طراحیشده برای ASGI | قابل استفاده با محدودیتهای معماری | پشتیبانی Async رو به گسترش |
| انتخاب مناسب | API و Microservice تایپمحور | سرویس کوچک با کنترل کامل | سامانه کامل با Admin و ORM |
اگر Admin Panel، Authentication یکپارچه و CMS میخواهید، Django ممکن است هزینه کلی را کمتر کند. اگر یک API مستقل با قرارداد OpenAPI، Validation قوی و معماری ASGI میخواهید، FastAPI انتخاب طبیعیتری است. انتخاب را بر اساس کل محصول انجام دهید، نه فقط Benchmark خام.
پیشنیازها و نصب FastAPI
برای پروژه جدید یک Virtual Environment جدا بسازید. مستندات فعلی FastAPI نصب بسته fastapi[standard] را پیشنهاد میکنند که CLI و Server لازم را نیز فراهم میکند. SQLAlchemy و Driver دیتابیس را جدا اضافه میکنیم.
python -m venv .venv# Windows PowerShell.venv\Scripts\Activate.ps1python -m pip install --upgrade pippip install "fastapi[standard]" sqlalchemyfastapi dev main.pySQLite برای یادگیری و تست محلی مناسب است. در Production معمولاً PostgreSQL یا دیتابیس مدیریتشده انتخاب میشود. Driver، Pool و تنظیم Connection باید متناسب با همان دیتابیس باشد؛ تنظیم SQLite را بدون بررسی به PostgreSQL منتقل نکنید.
ساخت Model دیتابیس و Schemaهای API
یکی از اشتباههای رایج این است که یک کلاس را هم Model دیتابیس، هم ورودی Create و هم خروجی عمومی در نظر بگیریم. این کار احتمال انتشار فیلدهای داخلی را بالا میبرد. در نمونه زیر Model ذخیرهسازی از Schemaهای Pydantic جدا است.
from sqlalchemy import Boolean, Stringfrom sqlalchemy.orm import DeclarativeBase, Mapped, mapped_columnfrom pydantic import BaseModel, ConfigDict, Field class Base(DeclarativeBase): pass class Task(Base): __tablename__ = "tasks" id: Mapped[int] = mapped_column(primary_key=True) title: Mapped[str] = mapped_column(String(120), index=True) is_done: Mapped[bool] = mapped_column(Boolean, default=False) class TaskCreate(BaseModel): title: str = Field(min_length=3, max_length=120) class TaskRead(BaseModel): model_config = ConfigDict(from_attributes=True) id: int title: str is_done: boolMapped و mapped_column سبک Declarative مدرن SQLAlchemy 2 هستند. محدودیت طول در Pydantic تجربه Client را بهتر میکند، ولی Constraintهای حیاتی باید در دیتابیس نیز تعریف شوند؛ Validation برنامه جای Unique Constraint یا Foreign Key را نمیگیرد.
Session دیتابیس را کوتاه و قابل پیشبینی نگه دارید
Session واحد سراسری برای همه درخواستها نسازید. هر Request باید Session خودش را دریافت و در پایان آزاد کند. این الگو با Dependency و yield قابل پیادهسازی است.
from collections.abc import Generatorfrom sqlalchemy import create_enginefrom sqlalchemy.orm import Session, sessionmaker engine = create_engine("sqlite:///./tasks.db")SessionLocal = sessionmaker(bind=engine, expire_on_commit=False) def get_session() -> Generator[Session, None, None]: with SessionLocal() as session: yield sessionدر Production، Connection String را از Secret Store یا Environment Variable بخوانید. مقدار Secret را Log نکنید و Credential را در Repository نگذارید. اندازه Pool باید با تعداد Workerها و سقف Connection دیتابیس هماهنگ باشد؛ چهار Worker با Pool دهتایی میتوانند چهل Connection بسازند.
ساخت Endpointهای CRUD با Status Code صحیح
from typing import Annotatedfrom fastapi import Depends, FastAPI, HTTPException, statusfrom sqlalchemy import selectfrom sqlalchemy.orm import Sessionfrom database import engine, get_sessionfrom models import Base, Task, TaskCreate, TaskRead Base.metadata.create_all(engine)app = FastAPI(title="Task API")DbSession = Annotated[Session, Depends(get_session)] @app.post("/tasks", response_model=TaskRead, status_code=status.HTTP_201_CREATED)def create_task(payload: TaskCreate, db: DbSession) -> Task: task = Task(title=payload.title) db.add(task) db.commit() db.refresh(task) return task @app.get("/tasks", response_model=list[TaskRead])def list_tasks(db: DbSession) -> list[Task]: statement = select(Task).order_by(Task.id.desc()).limit(100) return list(db.scalars(statement)) @app.patch("/tasks/{task_id}/done", response_model=TaskRead)def complete_task(task_id: int, db: DbSession) -> Task: task = db.get(Task, task_id) if task is None: raise HTTPException(status_code=404, detail="Task not found") task.is_done = True db.commit() db.refresh(task) return taskبرای Create پاسخ ۲۰۱ مناسب است؛ برای شناسه ناموجود ۴۰۴ برگردانید. خطای Validation معمولاً ۴۲۲ است. خطای داخلی را با Stack Trace به Client نفرستید. یک Error ID در پاسخ عمومی و جزئیات کامل فقط در Log امن، اشکالزدایی را بدون افشای اطلاعات ممکن میکند.
چرا create_all جای Migration نیست؟
create_all برای Demo جدول اولیه را میسازد، اما تغییر امن Schema در محیط واقعی به Migration نسخهدار نیاز دارد. ابزارهایی مانند Alembic کمک میکنند تغییر Column، Index و Constraint در Code Review دیده و در Staging آزموده شود. اجرای خودکار تغییر مخرب همزمان با Startup چند Replica میتواند Race Condition ایجاد کند.
قاعده عملی Migration
- Migration را جدا از Startup سرویس اجرا کنید.
- از الگوی Expand و Contract برای تغییر ناسازگار استفاده کنید.
- قبل از حذف Column، نسخههای قدیمی برنامه را از مدار خارج کنید.
- زمان Lock و حجم جدول را در Staging مشابه Production بسنجید.
- Backup قابل بازیابی و مسیر Rollback عملی داشته باشید.
Async را فقط وقتی زنجیره شما Async است انتخاب کنید
نوشتن async def بهتنهایی Query را غیرمسدودکننده نمیکند. اگر Driver و Session همگام باشند، عملیات دیتابیس همچنان Blocking است. برای مسیر Async باید Async Engine، Async Session و Driver سازگار داشته باشید. ترکیب تصادفی Sync و Async معمولاً پیچیدگی را زیاد میکند بیآنکه Throughput واقعی بهتر شود.
برای پروژه کوچک، نسخه Sync صحیح و قابل اندازهگیری انتخاب خوبی است. اگر بار I/O همزمان بالا دارید، ابتدا Profile بگیرید، سپس زنجیره کامل را Async کنید. کار CPU-heavy را نیز داخل Event Loop اجرا نکنید؛ آن را به Worker یا Queue بسپارید.
امنیت FastAPI در Production
- احراز هویت را با OAuth2/OIDC استاندارد یا Identity Provider معتبر انجام دهید.
- Password خام را ذخیره نکنید؛ Hash مقاوم و Salt مناسب لازم است.
- Authorization را در سطح Resource بررسی کنید، نه فقط وجود Token.
- Secret Key و Connection String را خارج از Source Control نگه دارید.
- CORS را فقط برای Originهای لازم باز کنید؛ CORS جای Authentication نیست.
- اندازه Body، زمان پاسخ و Rate Limit را در Reverse Proxy یا Gateway محدود کنید.
- Docs عمومی را در محیط حساس آگاهانه فعال یا محدود کنید.
- Dependencyها و Container Base Image را Patch و اسکن کنید.
مستندات رسمی FastAPI نمونه OAuth2 با Bearer و JWT دارد، اما آن نمونه نقطه شروع آموزشی است. در سامانه واقعی به Rotation کلید، Expiration کوتاه، Revocation یا Session Strategy، Audience و Issuer Validation و ثبت رویدادهای امنیتی نیاز دارید.
تستهایی که قبل از Deploy لازماند
| نوع تست | نمونه سناریو | ریسک پوششدادهشده |
|---|---|---|
| Validation | عنوان کوتاهتر از سه کاراکتر | ورودی نامعتبر |
| Integration | Create سپس Read از دیتابیس تست | ناسازگاری ORM و Schema |
| Authorization | کاربر A وظیفه کاربر B را تغییر دهد | IDOR و دسترسی افقی |
| Contract | OpenAPI با Client مصرفکننده مقایسه شود | شکستن قرارداد |
| Load | صدها Request همزمان روی Endpoint پرترافیک | Pool Exhaustion و Timeout |
| Migration | ارتقا از نسخه قبلی Schema روی داده واقعینما | Downtime و از دسترفتن داده |
در تست دیتابیس از Transaction یا Database مجزا استفاده کنید. SQLite in-memory همیشه رفتار PostgreSQL را بازنمایی نمیکند؛ Query، Constraint و Migrationهای حساس را روی Engine مشابه Production نیز اجرا کنید.
سناریوی Production: از درخواست تا دیتابیس
- Client درخواست HTTPS را به Load Balancer میفرستد.
- Gateway محدودیت Rate و اندازه Body را اعمال میکند.
- FastAPI داده را با Pydantic اعتبارسنجی میکند.
- Dependency هویت و مجوز Resource را بررسی میکند.
- Session کوتاهعمر Query پارامتری را اجرا میکند.
- Transaction Commit میشود یا در خطا Rollback میگردد.
- Response Model فقط فیلدهای مجاز را Serialize میکند.
- Trace، Metric و Log ساختاریافته بدون Secret ثبت میشوند.
برای هر Request یک Correlation ID داشته باشید تا مسیر آن بین Proxy، API و دیتابیس قابل ردیابی باشد. Metricهای نرخ خطا، Latency صدک ۹۵، تعداد Connection فعال و Queue Time از میانگین زمان پاسخ مفیدترند.
چکلیست استقرار
- تنظیمات Development، Staging و Production جدا هستند.
- Database Migration قبل از تغییر Traffic اجرا و بررسی میشود.
- Health Check سطح Process و Readiness سطح Dependency جداست.
- Timeout بین Client، Proxy، App و Database هماهنگ است.
- تعداد Workerها با CPU، Memory و Pool دیتابیس آزمایش شده است.
- Logها JSON و فاقد Token، Password و داده حساساند.
- Backup دیتابیس با آزمون Restore واقعی معتبر شده است.
- OpenAPI Contract نسخهبندی و Breaking Change کنترل میشود.
- Rollback نسخه برنامه و Migration مستند است.
- یک Smoke Test پس از Deploy، Create و Read اصلی را بررسی میکند.
خطاهای رایج
قرار دادن منطق تجاری در Route
Route باید ورودی HTTP را به Use Case تبدیل کند. اگر محاسبه قیمت، مجوز و چند Query در آن انباشته شود، تست و استفاده مجدد دشوار میشود. Service Layer کوچک و روشن بهتر از Route چندصدخطی است.
برگرداندن مستقیم Model داخلی
بدون Response Model ممکن است فیلد داخلی در خروجی ظاهر شود. Schema خروجی را صریح نگه دارید و برای نقشهای مختلف Schema جدا بسازید.
Query بدون Pagination
لیست نامحدود دیر یا زود Memory و Latency را خراب میکند. Limit سقفدار، Cursor یا Pagination پایدار و Index متناسب با Sort تعریف کنید.
نادیدهگرفتن Transaction
چند تغییر مرتبط باید اتمیک باشند. محل Commit را مشخص کنید، در شکست Rollback انجام دهید و برای عملیات قابل تکرار Idempotency Key در نظر بگیرید.
جمعبندی
FastAPI شروع سریعی دارد، اما کیفیت Production از مرزبندی مسئولیتها میآید: Pydantic قرارداد API را کنترل میکند، SQLAlchemy دسترسی داده را مدیریت میکند و Dependencyها منابعی مانند Session و هویت را به شکل قابل تست تزریق میکنند. Schema ورودی و خروجی را جدا نگه دارید، Migration را نسخهدار کنید و Async را فقط با زنجیره کاملاً سازگار به کار ببرید.
برای درک بهتر پایه زبان، ابتدا راهنمای برنامهنویسی پایتون را بخوانید. اگر میخواهید سرویس را همراه دیتابیس و Reverse Proxy اجرا کنید، راهنمای Docker Compose در Production ادامه مناسبی است.
برای یادگیری پروژهمحور پایتون و ساخت API حرفهای، دورهها و مسیرهای آموزشی خانه برنامهنویسان البرز را بررسی کنید.