وبلاگ / راهنمای مسیر یادگیری و فناوری
راهنمای مسیر یادگیری و فناوری

آموزش FastAPI و SQLAlchemy؛ ساخت API پایتون آماده Production

راهنمای پروژه‌محور ساخت REST API با FastAPI، Pydantic و SQLAlchemy؛ از نصب و مدل‌سازی تا Session دیتابیس، مدیریت خطا، امنیت، تست و استقرار Production.

· ۳۳ بازدید

یک Framework مدرن پایتون برای ساخت API است که از Type Hintهای استاندارد برای اعتبارسنجی ورودی، تولید Schema و مستندات OpenAPI استفاده می‌کند. اگر فقط یک Route نمونه بسازیم، شروع کار بسیار کوتاه است؛ اما API واقعی به ساختار پروژه، چرخه صحیح Session دیتابیس، مدل‌های ورودی و خروجی جدا، مدیریت خطا، تست، امنیت و استقرار قابل تکرار نیاز دارد.

در این آموزش یک API مدیریت وظایف می‌سازیم که Create، Read و Update دارد و از ، و 2 استفاده می‌کند. هدف کپی‌کردن چند خط کد نیست؛ هدف این است که بدانید هر لایه چه مسئولیتی دارد و نسخه Production چه تفاوتی با Demo دارد.

چرا اکنون ارزش یادگیری دارد؟

در بررسی واقعی Google Trends ایران در ۹ سپتامبر ۲۰۲۶، میانگین علاقه ۷۴ در بازه ۹۰روزه و ۲۲ در بازه ۱۲ماهه داشت؛ یعنی تقاضای پایه قوی است. در مقایسه مستقیم حجم کوچک‌تری داشت، اما Related Query عبارت در ۹۰ روز ۳۰۰٪ رشد کرد، به وضعیت Breakout رسید و documentation در ۱۲ ماه ۱۱۰٪ رشد داشت. عبارت web framework نیز Breakout بود.

پیشنهادهای واقعی Google Search شامل tutorial، ، documentation، install و مقایسه با Flask و Django بود. پس این موضوع یک «زیرموضوع رو‌به‌رشد» در اکوسیستم پرتقاضای است، نه صرفاً یک واژه پرحجم عمومی.

چه چیزی را خودکار می‌کند؟

  • تبدیل Path، Query، Header و Body به Typeهای پایتون
  • اعتبارسنجی داده با و پاسخ خطای ساختاریافته
  • تولید OpenAPI و رابط Swagger UI در مسیر /docs
  • Dependency Injection برای دیتابیس، احراز هویت و سرویس‌ها
  • پشتیبانی طبیعی از تابع‌های Sync و Async
  • تعریف Response Model برای کنترل خروجی عمومی

این قابلیت‌ها سرعت توسعه را بالا می‌برند، ولی جای طراحی Domain، Transaction، Authorization و Monitoring را نمی‌گیرند. Type Hint قرارداد ورودی را روشن می‌کند؛ تصمیم اینکه چه کسی اجازه تغییر یک Task را دارد هنوز مسئولیت برنامه شماست.

یا Flask یا Django؟

معیارFlaskDjango
تمرکز اصلیAPI مبتنی بر Type Hintهسته سبک و انعطاف‌پذیرFramework کامل وب
اعتبارسنجی و Schema خودکاربا Extension یا کد دستیForm و Serializer با ابزارهای مکمل
مستندات APIOpenAPI و Swagger خودکارنیازمند افزونهمعمولاً با Django REST Framework
ORM داخلیندارد؛ انتخاب آزادندارد؛ انتخاب آزادORM یکپارچه
Asyncطراحی‌شده برای ASGIقابل استفاده با محدودیت‌های معماریپشتیبانی Async رو به گسترش
انتخاب مناسبAPI و Microservice تایپ‌محورسرویس کوچک با کنترل کاملسامانه کامل با Admin و ORM

اگر Admin Panel، Authentication یکپارچه و CMS می‌خواهید، Django ممکن است هزینه کلی را کمتر کند. اگر یک API مستقل با قرارداد OpenAPI، Validation قوی و معماری ASGI می‌خواهید، انتخاب طبیعی‌تری است. انتخاب را بر اساس کل محصول انجام دهید، نه فقط Benchmark خام.

پیش‌نیازها و نصب

برای پروژه جدید یک Virtual Environment جدا بسازید. مستندات فعلی نصب بسته fastapi[standard] را پیشنهاد می‌کنند که CLI و Server لازم را نیز فراهم می‌کند. و Driver دیتابیس را جدا اضافه می‌کنیم.

terminal.shBash
python -m venv .venv# Windows PowerShell.venv\Scripts\Activate.ps1python -m pip install --upgrade pippip install "fastapi[standard]" sqlalchemyfastapi dev main.py
UTF-8Bash

SQLite برای یادگیری و تست محلی مناسب است. در Production معمولاً PostgreSQL یا دیتابیس مدیریت‌شده انتخاب می‌شود. Driver، Pool و تنظیم Connection باید متناسب با همان دیتابیس باشد؛ تنظیم SQLite را بدون بررسی به PostgreSQL منتقل نکنید.

ساخت Model دیتابیس و Schemaهای API

یکی از اشتباه‌های رایج این است که یک کلاس را هم Model دیتابیس، هم ورودی Create و هم خروجی عمومی در نظر بگیریم. این کار احتمال انتشار فیلدهای داخلی را بالا می‌برد. در نمونه زیر Model ذخیره‌سازی از Schemaهای جدا است.

models.py
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: bool
UTF-8

Mapped و mapped_column سبک Declarative مدرن 2 هستند. محدودیت طول در تجربه Client را بهتر می‌کند، ولی Constraintهای حیاتی باید در دیتابیس نیز تعریف شوند؛ Validation برنامه جای Unique Constraint یا Foreign Key را نمی‌گیرد.

Session دیتابیس را کوتاه و قابل پیش‌بینی نگه دارید

Session واحد سراسری برای همه درخواست‌ها نسازید. هر Request باید Session خودش را دریافت و در پایان آزاد کند. این الگو با Dependency و yield قابل پیاده‌سازی است.

database.py
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
UTF-8

در Production، Connection String را از Secret Store یا Environment Variable بخوانید. مقدار Secret را Log نکنید و Credential را در Repository نگذارید. اندازه Pool باید با تعداد Workerها و سقف Connection دیتابیس هماهنگ باشد؛ چهار Worker با Pool ده‌تایی می‌توانند چهل Connection بسازند.

ساخت Endpointهای CRUD با Status Code صحیح

main.py
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
UTF-8

برای 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

  1. Migration را جدا از Startup سرویس اجرا کنید.
  2. از الگوی Expand و Contract برای تغییر ناسازگار استفاده کنید.
  3. قبل از حذف Column، نسخه‌های قدیمی برنامه را از مدار خارج کنید.
  4. زمان Lock و حجم جدول را در Staging مشابه Production بسنجید.
  5. 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 بسپارید.

امنیت در 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 و اسکن کنید.

مستندات رسمی نمونه OAuth2 با Bearer و JWT دارد، اما آن نمونه نقطه شروع آموزشی است. در سامانه واقعی به Rotation کلید، Expiration کوتاه، Revocation یا Session Strategy، Audience و Issuer Validation و ثبت رویدادهای امنیتی نیاز دارید.

تست‌هایی که قبل از Deploy لازم‌اند

نوع تستنمونه سناریوریسک پوشش‌داده‌شده
Validationعنوان کوتاه‌تر از سه کاراکترورودی نامعتبر
IntegrationCreate سپس Read از دیتابیس تستناسازگاری ORM و Schema
Authorizationکاربر A وظیفه کاربر B را تغییر دهدIDOR و دسترسی افقی
ContractOpenAPI با Client مصرف‌کننده مقایسه شودشکستن قرارداد
Loadصدها Request هم‌زمان روی Endpoint پرترافیکPool Exhaustion و Timeout
Migrationارتقا از نسخه قبلی Schema روی داده واقعی‌نماDowntime و از دست‌رفتن داده

در تست دیتابیس از Transaction یا Database مجزا استفاده کنید. SQLite in-memory همیشه رفتار PostgreSQL را بازنمایی نمی‌کند؛ Query، Constraint و Migrationهای حساس را روی Engine مشابه Production نیز اجرا کنید.

سناریوی Production: از درخواست تا دیتابیس

  1. Client درخواست HTTPS را به Load Balancer می‌فرستد.
  2. Gateway محدودیت Rate و اندازه Body را اعمال می‌کند.
  3. داده را با اعتبارسنجی می‌کند.
  4. Dependency هویت و مجوز Resource را بررسی می‌کند.
  5. Session کوتاه‌عمر Query پارامتری را اجرا می‌کند.
  6. Transaction Commit می‌شود یا در خطا Rollback می‌گردد.
  7. Response Model فقط فیلدهای مجاز را Serialize می‌کند.
  8. 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 در نظر بگیرید.

جمع‌بندی

شروع سریعی دارد، اما کیفیت Production از مرزبندی مسئولیت‌ها می‌آید: قرارداد API را کنترل می‌کند، دسترسی داده را مدیریت می‌کند و Dependencyها منابعی مانند Session و هویت را به شکل قابل تست تزریق می‌کنند. Schema ورودی و خروجی را جدا نگه دارید، Migration را نسخه‌دار کنید و Async را فقط با زنجیره کاملاً سازگار به کار ببرید.

برای درک بهتر پایه زبان، ابتدا راهنمای برنامه‌نویسی پایتون را بخوانید. اگر می‌خواهید سرویس را همراه دیتابیس و Reverse Proxy اجرا کنید، راهنمای Docker Compose در Production ادامه مناسبی است.

برای یادگیری پروژه‌محور پایتون و ساخت API حرفه‌ای، دوره‌ها و مسیرهای آموزشی خانه برنامه‌نویسان البرز را بررسی کنید.