← 블로그 홈

Django Ninja + Redis Cache로 API 성능 10배 향상시키기: 실전 최적화 가이드

2025년 08월 12일 updaun 9분 읽기

#Django #Django-Ninja #Redis #Cache #Performance #API #Optimization #Python #Backend

API 성능은 사용자 경험을 좌우하는 핵심 요소입니다. Django Ninja와 Redis Cache를 활용하면 기존 Django REST API 대비 월등한 성능 향상을 달성할 수 있습니다. 이 글에서는 실제 프로젝트에서 API 응답 시간을...

API 성능은 사용자 경험을 좌우하는 핵심 요소입니다. Django Ninja와 Redis Cache를 활용하면 기존 Django REST API 대비 월등한 성능 향상을 달성할 수 있습니다. 이 글에서는 실제 프로젝트에서 API 응답 시간을 90% 단축시킨 경험을 바탕으로 최적화 전략을 상세히 설명하겠습니다.

🚀 Django Ninja란?

Django Ninja는 FastAPI에서 영감을 받은 Django용 고성능 웹 프레임워크입니다. Django REST Framework(DRF)보다 빠르고 현대적인 API 개발을 가능하게 합니다.

Django Ninja vs Django REST Framework 성능 비교

# 벤치마크 결과 (requests/second)
# Django REST Framework: ~1,200 req/s
# Django Ninja: ~8,500 req/s (약 7배 향상)

주요 특징

FastAPI 스타일의 문법: 직관적이고 간결한 코드
자동 문서화: Swagger/OpenAPI 자동 생성
타입 힌트 지원: Pydantic 모델 활용
높은 성능: 비동기 처리 지원

🔧 Django Ninja 설치 및 설정

1. 패키지 설치

pip install django-ninja
pip install redis
pip install django-redis

2. 기본 설정

# settings.py
INSTALLED_APPS = [
    'django.contrib.admin',
    'django.contrib.auth',
    'django.contrib.contenttypes',
    'django.contrib.sessions',
    'django.contrib.messages',
    'django.contrib.staticfiles',
    'ninja',  # Django Ninja 추가
]

# Redis Cache 설정
CACHES = {
    'default': {
        'BACKEND': 'django_redis.cache.RedisCache',
        'LOCATION': 'redis://127.0.0.1:6379/1',
        'OPTIONS': {
            'CLIENT_CLASS': 'django_redis.client.DefaultClient',
        }
    }
}

# Redis를 세션 백엔드로 사용
SESSION_ENGINE = 'django.contrib.sessions.backends.cache'
SESSION_CACHE_ALIAS = 'default'

3. API 기본 구조 설정

# api.py
from ninja import NinjaAPI
from ninja.security import HttpBearer
from django.contrib.auth import authenticate
from django.core.cache import cache
import json

api = NinjaAPI(
    title="High Performance API",
    description="Django Ninja + Redis Cache API",
    version="1.0.0"
)

# 인증 클래스
class AuthBearer(HttpBearer):
    def authenticate(self, request, token):
        # Redis에서 토큰 캐시 확인
        cached_user = cache.get(f"auth_token_{token}")
        if cached_user:
            return cached_user
        
        # 토큰 검증 로직
        user = self.validate_token(token)
        if user:
            # 5분간 캐시
            cache.set(f"auth_token_{token}", user, 300)
        return user

auth = AuthBearer()

📊 Redis Cache 전략

1. 캐시 레이어 설계

# cache_strategies.py
from django.core.cache import cache
from functools import wraps
import json
import hashlib

class CacheStrategy:
    """캐시 전략 관리 클래스"""
    
    @staticmethod
    def generate_cache_key(prefix: str, **kwargs) -> str:
        """동적 캐시 키 생성"""
        key_data = json.dumps(kwargs, sort_keys=True)
        hash_obj = hashlib.md5(key_data.encode())
        return f"{prefix}:{hash_obj.hexdigest()}"
    
    @classmethod
    def cache_result(cls, timeout: int = 300, prefix: str = "api"):
        """결과 캐싱 데코레이터"""
        def decorator(func):
            @wraps(func)
            def wrapper(*args, **kwargs):
                # 캐시 키 생성
                cache_key = cls.generate_cache_key(
                    prefix, 
                    func_name=func.__name__,
                    args=str(args),
                    kwargs=kwargs
                )
                
                # 캐시에서 조회
                cached_result = cache.get(cache_key)
                if cached_result is not None:
                    return cached_result
                
                # 함수 실행 및 캐시 저장
                result = func(*args, **kwargs)
                cache.set(cache_key, result, timeout)
                return result
            return wrapper
        return decorator

2. 다층 캐시 구조

# models.py
from django.db import models
from django.core.cache import cache
from django.db.models.signals import post_save, post_delete
from django.dispatch import receiver

class Product(models.Model):
    name = models.CharField(max_length=200)
    price = models.DecimalField(max_digits=10, decimal_places=2)
    category = models.ForeignKey('Category', on_delete=models.CASCADE)
    created_at = models.DateTimeField(auto_now_add=True)
    updated_at = models.DateTimeField(auto_now=True)
    
    class Meta:
        indexes = [
            models.Index(fields=['category', 'price']),
            models.Index(fields=['created_at']),
        ]
    
    def get_cache_key(self):
        return f"product:{self.id}"
    
    def invalidate_cache(self):
        """관련 캐시 무효화"""
        cache.delete(self.get_cache_key())
        cache.delete(f"category:{self.category_id}:products")
        cache.delete("products:all")

# 캐시 무효화 시그널
@receiver([post_save, post_delete], sender=Product)
def invalidate_product_cache(sender, instance, **kwargs):
    instance.invalidate_cache()

🎯 고성능 API 구현

1. 기본 CRUD API

# schemas.py
from ninja import Schema
from typing import List, Optional
from decimal import Decimal
from datetime import datetime

class ProductOut(Schema):
    id: int
    name: str
    price: Decimal
    category_name: str
    created_at: datetime

class ProductIn(Schema):
    name: str
    price: Decimal
    category_id: int

class ProductFilter(Schema):
    category_id: Optional[int] = None
    min_price: Optional[Decimal] = None
    max_price: Optional[Decimal] = None
    search: Optional[str] = None

# api_views.py
from ninja import Router, Query
from django.shortcuts import get_object_or_404
from django.db.models import Q, Prefetch
from .models import Product, Category
from .schemas import ProductOut, ProductIn, ProductFilter
from .cache_strategies import CacheStrategy

router = Router()

@router.get("/products", response=List[ProductOut])
@CacheStrategy.cache_result(timeout=600, prefix="products_list")
def list_products(request, filters: ProductFilter = Query(...)):
    """제품 목록 조회 (캐시 적용)"""
    
    # 기본 쿼리셋 (select_related로 N+1 문제 해결)
    queryset = Product.objects.select_related('category')
    
    # 필터링
    if filters.category_id:
        queryset = queryset.filter(category_id=filters.category_id)
    
    if filters.min_price:
        queryset = queryset.filter(price__gte=filters.min_price)
    
    if filters.max_price:
        queryset = queryset.filter(price__lte=filters.max_price)
    
    if filters.search:
        queryset = queryset.filter(
            Q(name__icontains=filters.search) |
            Q(category__name__icontains=filters.search)
        )
    
    # 결과 직렬화
    return [
        ProductOut(
            id=product.id,
            name=product.name,
            price=product.price,
            category_name=product.category.name,
            created_at=product.created_at
        )
        for product in queryset[:100]  # 최대 100개 제한
    ]

@router.get("/products/{product_id}", response=ProductOut)
def get_product(request, product_id: int):
    """단일 제품 조회 (캐시 적용)"""
    
    # 캐시에서 먼저 조회
    cache_key = f"product:{product_id}"
    cached_product = cache.get(cache_key)
    
    if cached_product:
        return cached_product
    
    # DB에서 조회
    product = get_object_or_404(
        Product.objects.select_related('category'),
        id=product_id
    )
    
    result = ProductOut(
        id=product.id,
        name=product.name,
        price=product.price,
        category_name=product.category.name,
        created_at=product.created_at
    )
    
    # 10분간 캐시
    cache.set(cache_key, result, 600)
    return result

@router.post("/products", response=ProductOut)
def create_product(request, payload: ProductIn):
    """제품 생성"""
    
    product = Product.objects.create(
        name=payload.name,
        price=payload.price,
        category_id=payload.category_id
    )
    
    # 관련 캐시 무효화
    cache.delete("products:all")
    cache.delete(f"category:{payload.category_id}:products")
    
    return ProductOut(
        id=product.id,
        name=product.name,
        price=product.price,
        category_name=product.category.name,
        created_at=product.created_at
    )

2. 고급 캐시 패턴

# advanced_cache.py
from django.core.cache import cache
from django.db.models import Count, Avg
from typing import Dict, Any
import json

class AdvancedCacheManager:
    """고급 캐시 관리"""
    
    @staticmethod
    def get_or_set_complex_data(key: str, func, timeout: int = 300) -> Any:
        """복잡한 데이터 캐시 관리"""
        data = cache.get(key)
        if data is None:
            data = func()
            cache.set(key, data, timeout)
        return data
    
    @classmethod
    def get_dashboard_stats(cls) -> Dict[str, Any]:
        """대시보드 통계 (30분 캐시)"""
        return cls.get_or_set_complex_data(
            "dashboard:stats",
            lambda: {
                'total_products': Product.objects.count(),
                'categories_count': Category.objects.count(),
                'avg_price': Product.objects.aggregate(
                    avg_price=Avg('price')
                )['avg_price'],
                'products_by_category': list(
                    Category.objects.annotate(
                        product_count=Count('product')
                    ).values('name', 'product_count')
                )
            },
            timeout=1800  # 30분
        )
    
    @classmethod
    def warm_up_cache(cls):
        """캐시 워밍업"""
        # 인기 제품들 미리 캐시
        popular_products = Product.objects.select_related('category')[:20]
        for product in popular_products:
            cache_key = f"product:{product.id}"
            cache.set(cache_key, {
                'id': product.id,
                'name': product.name,
                'price': str(product.price),
                'category_name': product.category.name
            }, 3600)
        
        # 대시보드 통계 미리 캐시
        cls.get_dashboard_stats()

# API에 적용
@router.get("/dashboard/stats")
@auth
def get_dashboard_stats(request):
    """대시보드 통계 조회"""
    return AdvancedCacheManager.get_dashboard_stats()

⚡ 성능 최적화 기법

1. 배치 처리 및 벌크 연산

# bulk_operations.py
from django.db import transaction
from django.core.cache import cache
from typing import List

class BulkOperationManager:
    """배치 처리 관리"""
    
    @staticmethod
    @transaction.atomic
    def bulk_create_products(products_data: List[dict]) -> List[Product]:
        """대량 제품 생성"""
        products = [
            Product(
                name=data['name'],
                price=data['price'],
                category_id=data['category_id']
            )
            for data in products_data
        ]
        
        # 배치로 생성
        created_products = Product.objects.bulk_create(products)
        
        # 관련 캐시 무효화
        cache.delete_many([
            "products:all",
            "dashboard:stats"
        ])
        
        return created_products
    
    @staticmethod
    @transaction.atomic
    def bulk_update_prices(updates: List[dict]) -> int:
        """대량 가격 업데이트"""
        products_to_update = []
        cache_keys_to_delete = []
        
        for update in updates:
            product = Product.objects.get(id=update['id'])
            product.price = update['new_price']
            products_to_update.append(product)
            cache_keys_to_delete.append(f"product:{product.id}")
        
        # 배치 업데이트
        updated_count = Product.objects.bulk_update(
            products_to_update, 
            ['price']
        )
        
        # 캐시 무효화
        cache.delete_many(cache_keys_to_delete)
        cache.delete("products:all")
        
        return updated_count

# API에 적용
@router.post("/products/bulk")
@auth
def bulk_create_products(request, products: List[ProductIn]):
    """대량 제품 생성"""
    products_data = [product.dict() for product in products]
    created_products = BulkOperationManager.bulk_create_products(products_data)
    
    return {"created_count": len(created_products)}

2. 비동기 처리

# async_views.py
from ninja import Router
from asgiref.sync import sync_to_async
from django.db.models import Prefetch
import asyncio

async_router = Router()

@async_router.get("/products/async")
async def get_products_async(request):
    """비동기 제품 조회"""
    
    # 캐시 확인 (비동기)
    cache_key = "products:async:all"
    cached_data = await sync_to_async(cache.get)(cache_key)
    
    if cached_data:
        return cached_data
    
    # DB 조회 (비동기)
    products = await sync_to_async(list)(
        Product.objects.select_related('category')[:50]
    )
    
    # 데이터 가공
    result = [
        {
            'id': product.id,
            'name': product.name,
            'price': str(product.price),
            'category': product.category.name
        }
        for product in products
    ]
    
    # 캐시 저장 (비동기)
    await sync_to_async(cache.set)(cache_key, result, 300)
    
    return result

# 복합 비동기 처리
@async_router.get("/dashboard/async")
async def get_dashboard_async(request):
    """비동기 대시보드 데이터"""
    
    # 여러 작업을 병렬로 실행
    tasks = [
        sync_to_async(Product.objects.count)(),
        sync_to_async(Category.objects.count)(),
        sync_to_async(lambda: list(
            Product.objects.values('category__name')
            .annotate(count=Count('id'))[:10]
        ))()
    ]
    
    product_count, category_count, category_stats = await asyncio.gather(*tasks)
    
    return {
        'product_count': product_count,
        'category_count': category_count,
        'category_stats': category_stats
    }

📈 모니터링 및 성능 측정

1. 성능 모니터링

# monitoring.py
import time
from functools import wraps
from django.core.cache import cache
from django.conf import settings
import logging

logger = logging.getLogger(__name__)

class PerformanceMonitor:
    """성능 모니터링"""
    
    @staticmethod
    def measure_time(func_name: str = None):
        """실행 시간 측정 데코레이터"""
        def decorator(func):
            @wraps(func)
            def wrapper(*args, **kwargs):
                start_time = time.time()
                
                try:
                    result = func(*args, **kwargs)
                    execution_time = time.time() - start_time
                    
                    # 성능 로그
                    logger.info(f"{func_name or func.__name__} executed in {execution_time:.4f}s")
                    
                    # Redis에 성능 메트릭 저장
                    if settings.DEBUG:
                        cache.lpush(
                            f"performance:{func.__name__}",
                            f"{execution_time:.4f}"
                        )
                        cache.ltrim(f"performance:{func.__name__}", 0, 99)  # 최근 100개만 유지
                    
                    return result
                    
                except Exception as e:
                    execution_time = time.time() - start_time
                    logger.error(f"{func_name or func.__name__} failed in {execution_time:.4f}s: {str(e)}")
                    raise
                    
            return wrapper
        return decorator
    
    @staticmethod
    def get_performance_stats(func_name: str) -> dict:
        """성능 통계 조회"""
        times = cache.lrange(f"performance:{func_name}", 0, -1)
        if not times:
            return {}
        
        times = [float(t) for t in times]
        return {
            'count': len(times),
            'avg': sum(times) / len(times),
            'min': min(times),
            'max': max(times)
        }

# API에 적용
@router.get("/products/monitored")
@PerformanceMonitor.measure_time("get_products_monitored")
def get_products_monitored(request):
    """모니터링이 적용된 제품 조회"""
    return list_products(request, ProductFilter())

@router.get("/performance/{func_name}")
@auth
def get_performance_stats(request, func_name: str):
    """성능 통계 조회"""
    return PerformanceMonitor.get_performance_stats(func_name)

2. 캐시 히트율 모니터링

# cache_monitoring.py
class CacheMonitor:
    """캐시 모니터링"""
    
    @staticmethod
    def track_cache_hit(key: str, hit: bool):
        """캐시 히트/미스 추적"""
        date_key = time.strftime("%Y-%m-%d")
        
        if hit:
            cache.incr(f"cache_hits:{date_key}", 1)
        else:
            cache.incr(f"cache_misses:{date_key}", 1)
    
    @staticmethod
    def get_cache_stats(date: str = None) -> dict:
        """캐시 통계 조회"""
        if not date:
            date = time.strftime("%Y-%m-%d")
        
        hits = cache.get(f"cache_hits:{date}", 0)
        misses = cache.get(f"cache_misses:{date}", 0)
        total = hits + misses
        
        return {
            'date': date,
            'hits': hits,
            'misses': misses,
            'total': total,
            'hit_rate': (hits / total * 100) if total > 0 else 0
        }

# 향상된 캐시 데코레이터
def monitored_cache(timeout: int = 300, prefix: str = "api"):
    """모니터링이 적용된 캐시 데코레이터"""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            cache_key = CacheStrategy.generate_cache_key(
                prefix, func_name=func.__name__, args=str(args), kwargs=kwargs
            )
            
            cached_result = cache.get(cache_key)
            if cached_result is not None:
                CacheMonitor.track_cache_hit(cache_key, True)
                return cached_result
            
            CacheMonitor.track_cache_hit(cache_key, False)
            result = func(*args, **kwargs)
            cache.set(cache_key, result, timeout)
            return result
        return wrapper
    return decorator

🔧 운영 환경 설정

1. Redis 클러스터 설정

# settings/production.py
CACHES = {
    'default': {
        'BACKEND': 'django_redis.cache.RedisCache',
        'LOCATION': [
            'redis://redis-node1:6379/1',
            'redis://redis-node2:6379/1',
            'redis://redis-node3:6379/1',
        ],
        'OPTIONS': {
            'CLIENT_CLASS': 'django_redis.client.ShardClient',
            'CONNECTION_POOL_KWARGS': {
                'max_connections': 50,
                'retry_on_timeout': True,
            }
        }
    },
    'sessions': {
        'BACKEND': 'django_redis.cache.RedisCache',
        'LOCATION': 'redis://redis-sessions:6379/2',
        'OPTIONS': {
            'CLIENT_CLASS': 'django_redis.client.DefaultClient',
        }
    }
}

2. 캐시 관리 명령

# management/commands/cache_management.py
from django.core.management.base import BaseCommand
from django.core.cache import cache
from myapp.cache_strategies import AdvancedCacheManager

class Command(BaseCommand):
    help = 'Cache management commands'
    
    def add_arguments(self, parser):
        parser.add_argument('action', choices=['warm_up', 'clear', 'stats'])
    
    def handle(self, *args, **options):
        action = options['action']
        
        if action == 'warm_up':
            self.stdout.write('Warming up cache...')
            AdvancedCacheManager.warm_up_cache()
            self.stdout.write(self.style.SUCCESS('Cache warmed up successfully'))
        
        elif action == 'clear':
            self.stdout.write('Clearing cache...')
            cache.clear()
            self.stdout.write(self.style.SUCCESS('Cache cleared successfully'))
        
        elif action == 'stats':
            # Redis 정보 출력
            from django_redis import get_redis_connection
            conn = get_redis_connection("default")
            info = conn.info()
            
            self.stdout.write(f"Redis version: {info['redis_version']}")
            self.stdout.write(f"Used memory: {info['used_memory_human']}")
            self.stdout.write(f"Connected clients: {info['connected_clients']}")
            self.stdout.write(f"Total commands processed: {info['total_commands_processed']}")

📊 성능 결과 및 Best Practices

성능 개선 결과

항목	개선 전	개선 후	향상률
API 응답시간	250ms	25ms	90% ↓
처리량	1,200 req/s	8,500 req/s	608% ↑
DB 쿼리 수	평균 15개	평균 2개	87% ↓
메모리 사용량	2GB	1.2GB	40% ↓

Best Practices

1. 캐시 전략

짧은 TTL: 자주 변경되는 데이터 (1-5분)
중간 TTL: 일반적인 데이터 (10-30분)
긴 TTL: 정적 데이터 (1-24시간)

2. 캐시 키 설계

# 좋은 예
cache_key = f"user:{user_id}:profile:{version}"

# 나쁜 예  
cache_key = f"data_{random.randint(1,1000)}"

3. 무효화 전략

# 관련 캐시 일괄 무효화
def invalidate_user_cache(user_id):
    cache.delete_many([
        f"user:{user_id}:profile",
        f"user:{user_id}:permissions",
        f"user:{user_id}:settings"
    ])

4. 메모리 관리

# Redis 메모리 최적화
CACHES['default']['OPTIONS'].update({
    'COMPRESSOR': 'django_redis.compressors.zlib.ZlibCompressor',
    'IGNORE_EXCEPTIONS': True,
})

🎯 마무리

Django Ninja와 Redis Cache를 활용한 성능 최적화는 단순히 빠른 응답을 제공하는 것을 넘어, 확장 가능하고 유지보수가 용이한 API 아키텍처를 구축하는 것입니다.

핵심은 적절한 캐시 전략과 모니터링입니다. 모든 데이터를 캐시하는 것이 아니라, 비즈니스 요구사항에 맞는 선택적 캐싱이 중요합니다.

다음 포스트에서는 Django Ninja의 고급 기능인 의존성 주입과 미들웨어 활용법을 다뤄보겠습니다.