پروژه نهایی – REST API با FastAPI

در این فصل پایانی، یک REST API کامل با FastAPI می‌سازیم – فریم‌ورک مدرن، سریع و async پایتون. شامل SQLAlchemy، Pydantic، احراز هویت JWT، تست خودکار، Docker و دیپلوی نهایی.

چرا FastAPI؟

• سریع: یکی از سریع‌ترین فریم‌ورک‌های پایتون (هم‌سطح Node و Go)
• Async: پشتیبانی native از async/await
• Type Hints: validation خودکار از روی type hints
• OpenAPI: مستندات خودکار Swagger UI
• Pydantic: validation قدرتمند داده

نصب و راه‌اندازی

mkdir fastapi-shop
cd fastapi-shop

# venv
python -m venv venv
source venv/bin/activate    # Linux/Mac
venvScriptsactivate       # Windows

# نصب
pip install fastapi uvicorn[standard]
pip install sqlalchemy alembic
pip install python-jose[cryptography] passlib[bcrypt]
pip install pydantic[email]
pip install pytest httpx pytest-asyncio

ساختار پروژه

fastapi-shop/
├── app/
│   ├── __init__.py
│   ├── main.py              # FastAPI app
│   ├── database.py          # SQLAlchemy setup
│   ├── models.py            # SQLAlchemy models
│   ├── schemas.py           # Pydantic schemas
│   ├── crud.py              # Database operations
│   ├── auth.py              # JWT authentication
│   ├── dependencies.py      # FastAPI dependencies
│   └── routers/
│       ├── __init__.py
│       ├── users.py
│       └── products.py
├── tests/
│   ├── __init__.py
│   ├── conftest.py
│   └── test_main.py
├── Dockerfile
├── docker-compose.yml
├── requirements.txt
└── .env

Hello FastAPI

# app/main.py
from fastapi import FastAPI

app = FastAPI(
    title="Shop API",
    description="API فروشگاه اینترنتی",
    version="1.0.0"
)

@app.get("/")
def root():
    return {"message": "خوش آمدید"}

@app.get("/health")
def health():
    return {"status": "ok"}

uvicorn app.main:app --reload

# حالا:
# http://localhost:8000        - API
# http://localhost:8000/docs   - Swagger UI خودکار
# http://localhost:8000/redoc  - ReDoc

دیتابیس با SQLAlchemy

# app/database.py
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

DATABASE_URL = "sqlite:///./shop.db"
# DATABASE_URL = "postgresql://user:pass@localhost/shop"  # برای Postgres

engine = create_engine(
    DATABASE_URL,
    connect_args={"check_same_thread": False}  # فقط برای SQLite
)

SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

def get_db():
    """Dependency برای دریافت session دیتابیس"""
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

Models

# app/models.py
from sqlalchemy import Column, Integer, String, Float, Boolean, ForeignKey, DateTime
from sqlalchemy.orm import relationship
from datetime import datetime
from .database import Base

class User(Base):
    __tablename__ = "users"
    
    id = Column(Integer, primary_key=True, index=True)
    username = Column(String(50), unique=True, index=True)
    email = Column(String(100), unique=True, index=True)
    hashed_password = Column(String(255))
    is_active = Column(Boolean, default=True)
    created_at = Column(DateTime, default=datetime.utcnow)
    
    products = relationship("Product", back_populates="owner")

class Product(Base):
    __tablename__ = "products"
    
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String(200), index=True)
    description = Column(String(1000))
    price = Column(Float)
    stock = Column(Integer, default=0)
    owner_id = Column(Integer, ForeignKey("users.id"))
    created_at = Column(DateTime, default=datetime.utcnow)
    
    owner = relationship("User", back_populates="products")

Pydantic Schemas

# app/schemas.py
from pydantic import BaseModel, EmailStr, Field
from datetime import datetime

# User
class UserBase(BaseModel):
    username: str = Field(..., min_length=3, max_length=50)
    email: EmailStr

class UserCreate(UserBase):
    password: str = Field(..., min_length=6)

class UserResponse(UserBase):
    id: int
    is_active: bool
    created_at: datetime
    
    class Config:
        from_attributes = True  # برای SQLAlchemy

# Product
class ProductBase(BaseModel):
    name: str = Field(..., min_length=1, max_length=200)
    description: str | None = None
    price: float = Field(..., gt=0)
    stock: int = Field(default=0, ge=0)

class ProductCreate(ProductBase):
    pass

class ProductResponse(ProductBase):
    id: int
    owner_id: int
    created_at: datetime
    
    class Config:
        from_attributes = True

# Token
class Token(BaseModel):
    access_token: str
    token_type: str

class TokenData(BaseModel):
    username: str | None = None

احراز هویت با JWT

# app/auth.py
from datetime import datetime, timedelta, timezone
from jose import JWTError, jwt
from passlib.context import CryptContext
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from sqlalchemy.orm import Session
from .database import get_db
from . import models, schemas

SECRET_KEY = "your-secret-key-change-this"  # از env بخوانید
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="login")

def hash_password(password: str) -> str:
    return pwd_context.hash(password)

def verify_password(plain: str, hashed: str) -> bool:
    return pwd_context.verify(plain, hashed)

def create_access_token(data: dict, expires_delta: timedelta | None = None):
    to_encode = data.copy()
    expire = datetime.now(timezone.utc) + (
        expires_delta or timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    )
    to_encode.update({"exp": expire})
    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)

async def get_current_user(
    token: str = Depends(oauth2_scheme),
    db: Session = Depends(get_db)
):
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="نمی‌توان اطلاعات هویت را تایید کرد",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username = payload.get("sub")
        if username is None:
            raise credentials_exception
    except JWTError:
        raise credentials_exception
    
    user = db.query(models.User).filter(models.User.username == username).first()
    if user is None:
        raise credentials_exception
    return user

CRUD Operations

# app/crud.py
from sqlalchemy.orm import Session
from . import models, schemas, auth

# Users
def get_user_by_username(db: Session, username: str):
    return db.query(models.User).filter(models.User.username == username).first()

def create_user(db: Session, user: schemas.UserCreate):
    hashed = auth.hash_password(user.password)
    db_user = models.User(
        username=user.username,
        email=user.email,
        hashed_password=hashed
    )
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return db_user

def authenticate_user(db: Session, username: str, password: str):
    user = get_user_by_username(db, username)
    if not user or not auth.verify_password(password, user.hashed_password):
        return None
    return user

# Products
def get_products(db: Session, skip: int = 0, limit: int = 100):
    return db.query(models.Product).offset(skip).limit(limit).all()

def get_product(db: Session, product_id: int):
    return db.query(models.Product).filter(models.Product.id == product_id).first()

def create_product(db: Session, product: schemas.ProductCreate, owner_id: int):
    db_product = models.Product(**product.model_dump(), owner_id=owner_id)
    db.add(db_product)
    db.commit()
    db.refresh(db_product)
    return db_product

def update_product(db: Session, product_id: int, product: schemas.ProductCreate):
    db_product = get_product(db, product_id)
    if db_product:
        for key, value in product.model_dump().items():
            setattr(db_product, key, value)
        db.commit()
        db.refresh(db_product)
    return db_product

def delete_product(db: Session, product_id: int):
    db_product = get_product(db, product_id)
    if db_product:
        db.delete(db_product)
        db.commit()
        return True
    return False

Routers

# app/routers/users.py
from fastapi import APIRouter, Depends, HTTPException
from fastapi.security import OAuth2PasswordRequestForm
from sqlalchemy.orm import Session
from .. import crud, schemas, auth
from ..database import get_db

router = APIRouter(prefix="/users", tags=["users"])

@router.post("/register", response_model=schemas.UserResponse)
def register(user: schemas.UserCreate, db: Session = Depends(get_db)):
    if crud.get_user_by_username(db, user.username):
        raise HTTPException(400, "این نام کاربری قبلاً ثبت شده")
    return crud.create_user(db, user)

@router.post("/login", response_model=schemas.Token)
def login(
    form_data: OAuth2PasswordRequestForm = Depends(),
    db: Session = Depends(get_db)
):
    user = crud.authenticate_user(db, form_data.username, form_data.password)
    if not user:
        raise HTTPException(401, "نام کاربری یا رمز اشتباه")
    
    token = auth.create_access_token(data={"sub": user.username})
    return {"access_token": token, "token_type": "bearer"}

@router.get("/me", response_model=schemas.UserResponse)
def me(current_user = Depends(auth.get_current_user)):
    return current_user

# app/routers/products.py
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.orm import Session
from typing import List
from .. import crud, schemas, auth
from ..database import get_db

router = APIRouter(prefix="/products", tags=["products"])

@router.get("/", response_model=List[schemas.ProductResponse])
def list_products(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    return crud.get_products(db, skip=skip, limit=limit)

@router.get("/{product_id}", response_model=schemas.ProductResponse)
def get_product(product_id: int, db: Session = Depends(get_db)):
    product = crud.get_product(db, product_id)
    if not product:
        raise HTTPException(404, "محصول یافت نشد")
    return product

@router.post("/", response_model=schemas.ProductResponse, status_code=201)
def create_product(
    product: schemas.ProductCreate,
    db: Session = Depends(get_db),
    current_user = Depends(auth.get_current_user)
):
    return crud.create_product(db, product, current_user.id)

@router.put("/{product_id}", response_model=schemas.ProductResponse)
def update_product(
    product_id: int,
    product: schemas.ProductCreate,
    db: Session = Depends(get_db),
    current_user = Depends(auth.get_current_user)
):
    db_product = crud.get_product(db, product_id)
    if not db_product:
        raise HTTPException(404, "محصول یافت نشد")
    if db_product.owner_id != current_user.id:
        raise HTTPException(403, "دسترسی ندارید")
    return crud.update_product(db, product_id, product)

@router.delete("/{product_id}", status_code=204)
def delete_product(
    product_id: int,
    db: Session = Depends(get_db),
    current_user = Depends(auth.get_current_user)
):
    db_product = crud.get_product(db, product_id)
    if not db_product:
        raise HTTPException(404, "محصول یافت نشد")
    if db_product.owner_id != current_user.id:
        raise HTTPException(403, "دسترسی ندارید")
    crud.delete_product(db, product_id)

main.py نهایی

# app/main.py
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from .database import engine, Base
from .routers import users, products

# ساخت جداول
Base.metadata.create_all(bind=engine)

app = FastAPI(
    title="Shop API",
    description="API فروشگاه اینترنتی - ICSD",
    version="1.0.0"
)

# CORS
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],  # در تولید مشخص کنید
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# Routers
app.include_router(users.router)
app.include_router(products.router)

@app.get("/")
def root():
    return {"message": "Shop API", "docs": "/docs"}

تست‌نویسی

# tests/conftest.py
import pytest
from fastapi.testclient import TestClient
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from app.main import app
from app.database import Base, get_db

SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"
engine = create_engine(SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False})
TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

@pytest.fixture
def db():
    Base.metadata.create_all(bind=engine)
    db = TestingSessionLocal()
    yield db
    db.close()
    Base.metadata.drop_all(bind=engine)

@pytest.fixture
def client(db):
    def override_get_db():
        yield db
    app.dependency_overrides[get_db] = override_get_db
    yield TestClient(app)
    app.dependency_overrides.clear()

# tests/test_main.py
def test_root(client):
    response = client.get("/")
    assert response.status_code == 200

def test_register(client):
    response = client.post("/users/register", json={
        "username": "testuser",
        "email": "test@example.com",
        "password": "secret123"
    })
    assert response.status_code == 200
    assert response.json()["username"] == "testuser"

def test_login_and_create_product(client):
    # ثبت‌نام
    client.post("/users/register", json={
        "username": "alice",
        "email": "alice@example.com",
        "password": "password"
    })
    
    # ورود
    response = client.post("/users/login", data={
        "username": "alice",
        "password": "password"
    })
    token = response.json()["access_token"]
    
    # ساخت محصول
    response = client.post(
        "/products/",
        json={"name": "فرش", "price": 5000000, "stock": 10},
        headers={"Authorization": f"Bearer {token}"}
    )
    assert response.status_code == 201
    assert response.json()["name"] == "فرش"

Docker

# Dockerfile
FROM python:3.12-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

EXPOSE 8000

CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]

# docker-compose.yml
version: "3.9"

services:
  api:
    build: .
    ports:
      - "8000:8000"
    environment:
      - DATABASE_URL=postgresql://shop:secret@db/shopdb
    depends_on:
      - db
  
  db:
    image: postgres:16
    environment:
      - POSTGRES_USER=shop
      - POSTGRES_PASSWORD=secret
      - POSTGRES_DB=shopdb
    volumes:
      - postgres_data:/var/lib/postgresql/data
    ports:
      - "5432:5432"

volumes:
  postgres_data:

docker-compose up -d
docker-compose logs -f api

دیپلوی

گزینه‌ها

• VPS (DigitalOcean، Hetzner): Docker + Nginx + Gunicorn
• Railway/Render: ساده، با Git push
• AWS/GCP: قدرتمند، پیچیده‌تر
• سرورهای ایرانی: لیارا، آروان، رابین

پروداکشن – Gunicorn + Uvicorn workers

pip install gunicorn

gunicorn app.main:app 
  -w 4 
  -k uvicorn.workers.UvicornWorker 
  --bind 0.0.0.0:8000

قدم‌های بعدی

• Migration با Alembic
• کش با Redis
• Background tasks با Celery یا BackgroundTasks
• WebSocket برای real-time
• Rate limiting با slowapi
• Logging با loguru یا structlog
• Monitoring با Prometheus + Grafana
• Database: PostgreSQL (دوره بعدی!)

جمع‌بندی – پایان دوره

تبریک! 🎉 شما این دوره را به پایان رساندید. در 15 فصل آموختیم:

• OOP پیشرفته و Magic Methods
• Decorators، Generators، Context Managers
• Type Hints و Static Typing
• Async/Await و Concurrency
• Metaclasses و Functional Programming
• Testing، Performance، Packaging
• Design Patterns و پروژه واقعی FastAPI

مرحله بعدی: تسلط بر دیتابیس‌ها با دوره PostgreSQL.