Python 백엔드 2025 완전 가이드: FastAPI vs Django, 비동기 처리, AI 서비스 구축까지

FastAPI를 선택하세요:
  - 새로운 API 서비스를 빌드할 때
  - AI/ML 모델 서빙이 필요할 때
  - 높은 동시성이 요구될 때
  - 마이크로서비스 아키텍처일 때

Django를 선택하세요:
  - 관리자 패널이 필요할 때
  - 빠른 CRUD 앱 개발이 필요할 때
  - 풀스택 웹 애플리케이션일 때
  - 팀에 Django 경험이 풍부할 때

Flask를 선택하세요:
  - 극도로 단순한 API일 때
  - 레거시 시스템 유지보수일 때
  - 학습 목적일 때

from fastapi import FastAPI, HTTPException, status
from pydantic import BaseModel, Field
from datetime import datetime

app = FastAPI(
    title="My API",
    version="1.0.0",
    docs_url="/docs",
    redoc_url="/redoc",
)

class UserCreate(BaseModel):
    username: str = Field(..., min_length=3, max_length=50)
    email: str = Field(..., pattern=r"^[\w\.\-]+@[\w\.\-]+\.\w+$")
    age: int = Field(..., ge=0, le=150)

class UserResponse(BaseModel):
    id: int
    username: str
    email: str
    created_at: datetime

    model_config = {"from_attributes": True}

@app.post("/users", response_model=UserResponse, status_code=status.HTTP_201_CREATED)
async def create_user(user: UserCreate):
    # 비즈니스 로직
    return UserResponse(
        id=1,
        username=user.username,
        email=user.email,
        created_at=datetime.now(),
    )

@app.get("/users/{user_id}", response_model=UserResponse)
async def get_user(user_id: int):
    if user_id <= 0:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="User not found",
        )
    return UserResponse(id=user_id, username="test", email="t@t.com", created_at=datetime.now())

from pydantic import BaseModel, Field, field_validator, model_validator
from typing import Optional
from enum import Enum

class Role(str, Enum):
    ADMIN = "admin"
    USER = "user"
    MODERATOR = "moderator"

class UserProfile(BaseModel):
    name: str = Field(..., min_length=1, max_length=100)
    role: Role = Role.USER
    tags: list[str] = Field(default_factory=list, max_length=10)
    metadata: Optional[dict[str, str]] = None

    @field_validator("name")
    @classmethod
    def name_must_not_be_empty(cls, v: str) -> str:
        if not v.strip():
            raise ValueError("Name cannot be whitespace only")
        return v.strip()

    @model_validator(mode="after")
    def validate_admin_tags(self):
        if self.role == Role.ADMIN and not self.tags:
            raise ValueError("Admin must have at least one tag")
        return self

    model_config = {
        "json_schema_extra": {
            "examples": [
                {"name": "Alice", "role": "admin", "tags": ["ops"]}
            ]
        }
    }

from fastapi import Depends, Header, HTTPException
from sqlalchemy.ext.asyncio import AsyncSession
from typing import Annotated

# DB 세션 의존성
async def get_db() -> AsyncSession:
    async with async_session_maker() as session:
        try:
            yield session
        finally:
            await session.close()

# 인증 의존성
async def get_current_user(
    authorization: str = Header(...),
    db: AsyncSession = Depends(get_db),
) -> User:
    token = authorization.replace("Bearer ", "")
    user = await verify_token(token, db)
    if not user:
        raise HTTPException(status_code=401, detail="Invalid token")
    return user

# 권한 검증 의존성
def require_role(required_role: str):
    async def role_checker(user: User = Depends(get_current_user)):
        if user.role != required_role:
            raise HTTPException(status_code=403, detail="Insufficient permissions")
        return user
    return role_checker

# 사용
@app.get("/admin/users")
async def list_users(
    admin: Annotated[User, Depends(require_role("admin"))],
    db: Annotated[AsyncSession, Depends(get_db)],
):
    return await get_all_users(db)

from fastapi import Request
from fastapi.middleware.cors import CORSMiddleware
from starlette.middleware.base import BaseHTTPMiddleware
import time
import logging

logger = logging.getLogger(__name__)

# CORS 미들웨어
app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://myapp.com"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# 커스텀 미들웨어 — 요청 로깅
class RequestLoggingMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        start = time.perf_counter()
        response = await call_next(request)
        elapsed = time.perf_counter() - start
        logger.info(
            f"method={request.method} path={request.url.path} "
            f"status={response.status_code} duration={elapsed:.3f}s"
        )
        return response

app.add_middleware(RequestLoggingMiddleware)

# 글로벌 예외 핸들러
@app.exception_handler(ValueError)
async def value_error_handler(request: Request, exc: ValueError):
    return JSONResponse(
        status_code=422,
        content={"detail": str(exc), "type": "validation_error"},
    )

from fastapi import WebSocket, WebSocketDisconnect, BackgroundTasks

# WebSocket 채팅
class ConnectionManager:
    def __init__(self):
        self.active: list[WebSocket] = []

    async def connect(self, ws: WebSocket):
        await ws.accept()
        self.active.append(ws)

    def disconnect(self, ws: WebSocket):
        self.active.remove(ws)

    async def broadcast(self, message: str):
        for conn in self.active:
            await conn.send_text(message)

manager = ConnectionManager()

@app.websocket("/ws/chat")
async def websocket_chat(ws: WebSocket):
    await manager.connect(ws)
    try:
        while True:
            data = await ws.receive_text()
            await manager.broadcast(f"User: {data}")
    except WebSocketDisconnect:
        manager.disconnect(ws)

# 백그라운드 태스크
async def send_notification(email: str, message: str):
    # 이메일 발송 로직
    await asyncio.sleep(2)  # 시뮬레이션
    logger.info(f"Sent notification to {email}")

@app.post("/orders")
async def create_order(order: OrderCreate, background_tasks: BackgroundTasks):
    result = await save_order(order)
    background_tasks.add_task(send_notification, order.email, "Order confirmed!")
    return result

# models.py
from django.db import models
from django.core.validators import MinValueValidator

class Category(models.Model):
    name = models.CharField(max_length=100, unique=True)
    slug = models.SlugField(unique=True)

    class Meta:
        verbose_name_plural = "categories"
        ordering = ["name"]

    def __str__(self):
        return self.name

class Product(models.Model):
    name = models.CharField(max_length=200)
    category = models.ForeignKey(Category, on_delete=models.CASCADE, related_name="products")
    price = models.DecimalField(max_digits=10, decimal_places=2, validators=[MinValueValidator(0)])
    stock = models.PositiveIntegerField(default=0)
    is_active = models.BooleanField(default=True)
    created_at = models.DateTimeField(auto_now_add=True)
    updated_at = models.DateTimeField(auto_now=True)

    class Meta:
        indexes = [
            models.Index(fields=["category", "-created_at"]),
            models.Index(fields=["price"]),
        ]

    def __str__(self):
        return self.name

# 쿼리 예시
# N+1 문제 해결
products = Product.objects.select_related("category").filter(is_active=True)

# 집계
from django.db.models import Avg, Count, Q
stats = Category.objects.annotate(
    product_count=Count("products"),
    avg_price=Avg("products__price"),
    active_count=Count("products", filter=Q(products__is_active=True)),
)

# admin.py
from django.contrib import admin
from .models import Product, Category

@admin.register(Product)
class ProductAdmin(admin.ModelAdmin):
    list_display = ["name", "category", "price", "stock", "is_active", "created_at"]
    list_filter = ["is_active", "category", "created_at"]
    search_fields = ["name", "category__name"]
    list_editable = ["price", "stock", "is_active"]
    readonly_fields = ["created_at", "updated_at"]
    list_per_page = 25

    actions = ["activate_products", "deactivate_products"]

    @admin.action(description="Activate selected products")
    def activate_products(self, request, queryset):
        updated = queryset.update(is_active=True)
        self.message_user(request, f"{updated} products activated.")

    @admin.action(description="Deactivate selected products")
    def deactivate_products(self, request, queryset):
        updated = queryset.update(is_active=False)
        self.message_user(request, f"{updated} products deactivated.")

# serializers.py
from rest_framework import serializers
from .models import Product, Category

class CategorySerializer(serializers.ModelSerializer):
    product_count = serializers.IntegerField(read_only=True)

    class Meta:
        model = Category
        fields = ["id", "name", "slug", "product_count"]

class ProductSerializer(serializers.ModelSerializer):
    category_name = serializers.CharField(source="category.name", read_only=True)

    class Meta:
        model = Product
        fields = ["id", "name", "category", "category_name", "price", "stock", "is_active"]

# views.py
from rest_framework import viewsets, filters
from django_filters.rest_framework import DjangoFilterBackend

class ProductViewSet(viewsets.ModelViewSet):
    queryset = Product.objects.select_related("category").filter(is_active=True)
    serializer_class = ProductSerializer
    filter_backends = [DjangoFilterBackend, filters.SearchFilter, filters.OrderingFilter]
    filterset_fields = ["category", "is_active"]
    search_fields = ["name"]
    ordering_fields = ["price", "created_at"]
    ordering = ["-created_at"]

# Django 5.0+ async views
from django.http import JsonResponse
from asgiref.sync import sync_to_async

async def async_product_list(request):
    # ORM 호출은 sync_to_async로 감싸야 함
    products = await sync_to_async(list)(
        Product.objects.select_related("category").filter(is_active=True)[:20]
    )
    data = [
        {"id": p.id, "name": p.name, "price": str(p.price)}
        for p in products
    ]
    return JsonResponse({"products": data})

# Django Channels — WebSocket
# consumers.py
from channels.generic.websocket import AsyncJsonWebSocketConsumer

class ChatConsumer(AsyncJsonWebSocketConsumer):
    async def connect(self):
        self.room = self.scope["url_route"]["kwargs"]["room"]
        await self.channel_layer.group_add(self.room, self.channel_name)
        await self.accept()

    async def disconnect(self, close_code):
        await self.channel_layer.group_discard(self.room, self.channel_name)

    async def receive_json(self, content):
        await self.channel_layer.group_send(
            self.room,
            {"type": "chat.message", "message": content["message"]},
        )

    async def chat_message(self, event):
        await self.send_json({"message": event["message"]})

from sqlalchemy import String, Integer, ForeignKey, DateTime, func
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
from datetime import datetime

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(primary_key=True)
    username: Mapped[str] = mapped_column(String(50), unique=True, index=True)
    email: Mapped[str] = mapped_column(String(255), unique=True)
    created_at: Mapped[datetime] = mapped_column(DateTime, server_default=func.now())
    posts: Mapped[list["Post"]] = relationship(back_populates="author", lazy="selectin")

class Post(Base):
    __tablename__ = "posts"

    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(200))
    content: Mapped[str] = mapped_column(String)
    author_id: Mapped[int] = mapped_column(ForeignKey("users.id"))
    author: Mapped["User"] = relationship(back_populates="posts")

# 비동기 엔진 설정
engine = create_async_engine(
    "postgresql+asyncpg://user:pass@localhost/db",
    pool_size=20,
    max_overflow=10,
    pool_pre_ping=True,
    echo=False,
)

async_session = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)

from sqlalchemy import select, update, delete
from sqlalchemy.ext.asyncio import AsyncSession

class UserRepository:
    def __init__(self, session: AsyncSession):
        self.session = session

    async def get_by_id(self, user_id: int) -> User | None:
        result = await self.session.execute(
            select(User).where(User.id == user_id)
        )
        return result.scalar_one_or_none()

    async def get_by_username(self, username: str) -> User | None:
        result = await self.session.execute(
            select(User).where(User.username == username)
        )
        return result.scalar_one_or_none()

    async def list_with_posts(self, skip: int = 0, limit: int = 20) -> list[User]:
        result = await self.session.execute(
            select(User)
            .options(selectinload(User.posts))
            .offset(skip)
            .limit(limit)
            .order_by(User.created_at.desc())
        )
        return list(result.scalars().all())

    async def create(self, username: str, email: str) -> User:
        user = User(username=username, email=email)
        self.session.add(user)
        await self.session.commit()
        await self.session.refresh(user)
        return user

    async def update_email(self, user_id: int, new_email: str) -> bool:
        result = await self.session.execute(
            update(User).where(User.id == user_id).values(email=new_email)
        )
        await self.session.commit()
        return result.rowcount > 0

# 초기화
alembic init alembic

# 마이그레이션 생성
alembic revision --autogenerate -m "add users table"

# 마이그레이션 적용
alembic upgrade head

# 롤백
alembic downgrade -1

# alembic/env.py — 비동기 설정
from sqlalchemy.ext.asyncio import create_async_engine

def run_migrations_online():
    connectable = create_async_engine(get_url())

    async def do_run():
        async with connectable.connect() as connection:
            await connection.run_sync(do_run_migrations)

    asyncio.run(do_run())

import asyncio
import aiohttp

# 여러 API 동시 호출
async def fetch_url(session: aiohttp.ClientSession, url: str) -> dict:
    async with session.get(url) as response:
        return await response.json()

async def fetch_all(urls: list[str]) -> list[dict]:
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_url(session, url) for url in urls]
        return await asyncio.gather(*tasks, return_exceptions=True)

# 실행
urls = [
    "https://api.example.com/users",
    "https://api.example.com/products",
    "https://api.example.com/orders",
]
results = asyncio.run(fetch_all(urls))

import asyncio

# 동시에 최대 10개만 실행
semaphore = asyncio.Semaphore(10)

async def rate_limited_fetch(session, url):
    async with semaphore:
        async with session.get(url) as response:
            return await response.json()

async def fetch_many(urls: list[str]):
    async with aiohttp.ClientSession() as session:
        tasks = [rate_limited_fetch(session, url) for url in urls]
        return await asyncio.gather(*tasks)

비동기가 좋은 경우:
  - 외부 API 호출이 많은 경우
  - 데이터베이스 I/O가 빈번한 경우
  - WebSocket / 실시간 통신
  - 파일 I/O가 많은 경우

동기가 나은 경우:
  - CPU 집약적 작업 (이미지 처리, ML 추론)
  - 간단한 CRUD 작업
  - 레거시 라이브러리가 비동기를 지원하지 않는 경우
  - concurrent.futures.ProcessPoolExecutor 사용이 나은 경우

from concurrent.futures import ProcessPoolExecutor
import asyncio

def cpu_heavy_task(data: bytes) -> bytes:
    # CPU 집약적 작업 (이미지 리사이징 등)
    import hashlib
    return hashlib.sha256(data).digest()

async def process_files(files: list[bytes]) -> list[bytes]:
    loop = asyncio.get_event_loop()
    with ProcessPoolExecutor(max_workers=4) as executor:
        tasks = [
            loop.run_in_executor(executor, cpu_heavy_task, f)
            for f in files
        ]
        return await asyncio.gather(*tasks)

# celery_app.py
from celery import Celery

app = Celery(
    "myapp",
    broker="redis://localhost:6379/0",
    backend="redis://localhost:6379/1",
)

app.conf.update(
    task_serializer="json",
    accept_content=["json"],
    result_serializer="json",
    timezone="Asia/Seoul",
    enable_utc=True,
    task_track_started=True,
    task_acks_late=True,        # 작업 완료 후 ACK
    worker_prefetch_multiplier=1, # 한 번에 하나씩 가져옴
    task_reject_on_worker_lost=True,
)

from celery import shared_task
from celery.utils.log import get_task_logger

logger = get_task_logger(__name__)

@shared_task(
    bind=True,
    max_retries=3,
    default_retry_delay=60,
    autoretry_for=(ConnectionError, TimeoutError),
    retry_backoff=True,
    retry_jitter=True,
)
def send_email_task(self, to: str, subject: str, body: str):
    try:
        send_email(to, subject, body)
        logger.info(f"Email sent to {to}")
    except Exception as exc:
        logger.error(f"Email failed: {exc}")
        raise self.retry(exc=exc)

@shared_task(bind=True)
def process_ai_inference(self, model_name: str, input_data: dict):
    self.update_state(state="PROCESSING", meta={"progress": 0})
    # 모델 로딩
    model = load_model(model_name)
    self.update_state(state="PROCESSING", meta={"progress": 50})
    # 추론 실행
    result = model.predict(input_data)
    self.update_state(state="PROCESSING", meta={"progress": 100})
    return result

from celery.schedules import crontab

app.conf.beat_schedule = {
    "cleanup-expired-sessions": {
        "task": "tasks.cleanup_sessions",
        "schedule": crontab(hour=2, minute=0),  # 매일 새벽 2시
    },
    "sync-external-data": {
        "task": "tasks.sync_data",
        "schedule": 300.0,  # 5분마다
    },
    "weekly-report": {
        "task": "tasks.generate_report",
        "schedule": crontab(hour=9, minute=0, day_of_week="mon"),
    },
}

# Flower 실행
celery -A myapp flower --port=5555

# Flower 기능:
# - 실시간 워커 상태 모니터링
# - 태스크 진행 상황 추적
# - 실패한 태스크 재시도
# - 워커 풀 크기 조절

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate
from langchain.schema import StrOutputParser
import asyncio

app = FastAPI()

llm = ChatOpenAI(model="gpt-4o", temperature=0.7, streaming=True)

prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful assistant. Answer in Korean."),
    ("human", "{question}"),
])

chain = prompt | llm | StrOutputParser()

@app.post("/chat")
async def chat(question: str):
    response = await chain.ainvoke({"question": question})
    return {"answer": response}

@app.post("/chat/stream")
async def chat_stream(question: str):
    async def generate():
        async for chunk in chain.astream({"question": question}):
            yield f"data: {chunk}\n\n"
        yield "data: [DONE]\n\n"

    return StreamingResponse(generate(), media_type="text/event-stream")

from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings
from langchain.text_splitter import RecursiveCharacterTextSplitter

# 벡터 스토어 초기화
embeddings = OpenAIEmbeddings()
vectorstore = Chroma(persist_directory="./chroma_db", embedding_function=embeddings)
retriever = vectorstore.as_retriever(search_kwargs={"k": 5})

rag_prompt = ChatPromptTemplate.from_messages([
    ("system", "Answer based on the following context:\n{context}"),
    ("human", "{question}"),
])

@app.post("/rag/query")
async def rag_query(question: str):
    # 관련 문서 검색
    docs = await retriever.ainvoke(question)
    context = "\n".join([doc.page_content for doc in docs])

    # LLM으로 답변 생성
    response = await (rag_prompt | llm | StrOutputParser()).ainvoke(
        {"context": context, "question": question}
    )

    return {
        "answer": response,
        "sources": [{"content": d.page_content, "metadata": d.metadata} for d in docs],
    }

from contextlib import asynccontextmanager
import torch

# 앱 시작 시 모델 로딩 (lifespan)
ml_models = {}

@asynccontextmanager
async def lifespan(app: FastAPI):
    # 시작 시 모델 로딩
    ml_models["sentiment"] = load_sentiment_model()
    ml_models["ner"] = load_ner_model()
    yield
    # 종료 시 정리
    ml_models.clear()
    torch.cuda.empty_cache()

app = FastAPI(lifespan=lifespan)

@app.post("/predict/sentiment")
async def predict_sentiment(text: str):
    model = ml_models["sentiment"]
    # CPU 바운드 작업은 스레드 풀에서 실행
    loop = asyncio.get_event_loop()
    result = await loop.run_in_executor(None, model.predict, text)
    return {"sentiment": result}

import pytest
from httpx import AsyncClient, ASGITransport
from main import app

@pytest.fixture
async def client():
    transport = ASGITransport(app=app)
    async with AsyncClient(transport=transport, base_url="http://test") as ac:
        yield ac

@pytest.mark.anyio
async def test_create_user(client: AsyncClient):
    response = await client.post("/users", json={
        "username": "testuser",
        "email": "test@example.com",
        "age": 25,
    })
    assert response.status_code == 201
    data = response.json()
    assert data["username"] == "testuser"

@pytest.mark.anyio
async def test_get_user_not_found(client: AsyncClient):
    response = await client.get("/users/0")
    assert response.status_code == 404

@pytest.mark.anyio
async def test_create_user_invalid_email(client: AsyncClient):
    response = await client.post("/users", json={
        "username": "testuser",
        "email": "invalid-email",
        "age": 25,
    })
    assert response.status_code == 422

import factory
from factory.alchemy import SQLAlchemyModelFactory

class UserFactory(SQLAlchemyModelFactory):
    class Meta:
        model = User
        sqlalchemy_session_persistence = "commit"

    username = factory.Sequence(lambda n: f"user_{n}")
    email = factory.LazyAttribute(lambda o: f"{o.username}@example.com")

class PostFactory(SQLAlchemyModelFactory):
    class Meta:
        model = Post
        sqlalchemy_session_persistence = "commit"

    title = factory.Faker("sentence")
    content = factory.Faker("paragraph")
    author = factory.SubFactory(UserFactory)

# 사용
async def test_list_users(db_session):
    users = [UserFactory(session=db_session) for _ in range(10)]
    # 테스트 로직

# pyproject.toml
[tool.pytest.ini_options]
testpaths = ["tests"]
asyncio_mode = "auto"

[tool.coverage.run]
source = ["app"]
omit = ["tests/*", "alembic/*"]

[tool.coverage.report]
fail_under = 80
show_missing = true

# 실행
pytest --cov=app --cov-report=html tests/

# 개발
uvicorn main:app --reload --host 0.0.0.0 --port 8000

# 프로덕션 — Gunicorn + Uvicorn Workers
gunicorn main:app \
  -w 4 \
  -k uvicorn.workers.UvicornWorker \
  --bind 0.0.0.0:8000 \
  --timeout 120 \
  --graceful-timeout 30 \
  --access-logfile - \
  --error-logfile -

# Stage 1: 빌드
FROM python:3.12-slim AS builder

WORKDIR /app
RUN pip install --no-cache-dir poetry

COPY pyproject.toml poetry.lock ./
RUN poetry export -f requirements.txt --output requirements.txt --without-hashes
RUN pip install --no-cache-dir --prefix=/install -r requirements.txt

# Stage 2: 실행
FROM python:3.12-slim

WORKDIR /app
COPY --from=builder /install /usr/local
COPY . .

# 비 root 사용자
RUN adduser --disabled-password --gecos "" appuser
USER appuser

EXPOSE 8000

HEALTHCHECK --interval=30s --timeout=3s --start-period=5s --retries=3 \
  CMD python -c "import httpx; httpx.get('http://localhost:8000/health')"

CMD ["gunicorn", "main:app", "-w", "4", "-k", "uvicorn.workers.UvicornWorker", "--bind", "0.0.0.0:8000"]

import structlog
import logging

structlog.configure(
    processors=[
        structlog.contextvars.merge_contextvars,
        structlog.processors.add_log_level,
        structlog.processors.TimeStamper(fmt="iso"),
        structlog.processors.StackInfoRenderer(),
        structlog.processors.format_exc_info,
        structlog.processors.JSONRenderer(),
    ],
    wrapper_class=structlog.make_filtering_bound_logger(logging.INFO),
    logger_factory=structlog.PrintLoggerFactory(),
)

logger = structlog.get_logger()

# 사용
async def create_order(order: OrderCreate):
    logger.info("order_created", order_id=order.id, user_id=order.user_id, amount=order.total)

from fastapi import FastAPI
from datetime import datetime

@app.get("/health")
async def health_check():
    return {
        "status": "healthy",
        "timestamp": datetime.utcnow().isoformat(),
        "version": "1.0.0",
    }

@app.get("/health/ready")
async def readiness_check(db: AsyncSession = Depends(get_db)):
    try:
        await db.execute(text("SELECT 1"))
        db_status = "connected"
    except Exception:
        db_status = "disconnected"
        raise HTTPException(status_code=503)

    return {"status": "ready", "database": db_status}