Django Advanced Optimization – সম্পূর্ণ বাংলা গাইড

# ❌ খারাপ - N+1 queries
def get_products_with_category(request):
    products = Product.objects.all()  # 1 query
    for product in products:
        print(product.category.name)  # N queries (প্রতিটি product এর জন্য)

# ✅ ভালো - শুধু 2 queries
def get_products_with_category(request):
    products = Product.objects.select_related('category')\
                              .prefetch_related('tags')\
                              .all()
    for product in products:
        print(product.category.name)  # কোন extra query নেই

from django.db import models
from django.contrib.auth.models import User

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

    class Meta:
        verbose_name_plural = "Categories"

    def __str__(self):
        return self.name


class Tag(models.Model):
    name = models.CharField(max_length=50)

    def __str__(self):
        return self.name


class Product(models.Model):
    name = models.CharField(max_length=200, db_index=True)
    description = models.TextField()
    price = models.DecimalField(max_digits=10, decimal_places=2, db_index=True)
    stock = models.IntegerField(default=0)

    # Relations
    category = models.ForeignKey(Category, on_delete=models.CASCADE, related_name='products')
    tags = models.ManyToManyField(Tag, related_name='products', blank=True)
    created_by = models.ForeignKey(User, on_delete=models.SET_NULL, null=True)

    created_at = models.DateTimeField(auto_now_add=True, db_index=True)
    updated_at = models.DateTimeField(auto_now=True)
    is_active = models.BooleanField(default=True, db_index=True)

    class Meta:
        ordering = ['-created_at']
        indexes = [
            models.Index(fields=['name', 'price']),
            models.Index(fields=['category', 'is_active']),
            models.Index(fields=['-created_at']),
        ]

    def __str__(self):
        return self.name


class Review(models.Model):
    product = models.ForeignKey(Product, on_delete=models.CASCADE, related_name='reviews')
    user = models.ForeignKey(User, on_delete=models.CASCADE)
    rating = models.IntegerField(choices=[(i, i) for i in range(1, 6)])
    comment = models.TextField()
    created_at = models.DateTimeField(auto_now_add=True)

    class Meta:
        ordering = ['-created_at']
        unique_together = ['product', 'user']

from django.db.models import Prefetch, Count, Avg
from rest_framework import viewsets
from rest_framework.decorators import action
from rest_framework.response import Response

class OptimizedProductViewSet(viewsets.ModelViewSet):
    serializer_class = ProductSerializer
    permission_classes = [IsAuthenticated]

    def get_queryset(self):
        """
        Highly optimized queryset
        """
        return Product.objects.select_related(
            'category',
            'created_by'
        ).prefetch_related(
            'tags',
            Prefetch(
                'reviews',
                queryset=Review.objects.select_related('user').order_by('-created_at')[:5]
            )
        ).annotate(
            review_count=Count('reviews'),
            avg_rating=Avg('reviews__rating')
        ).only(
            'id', 'name', 'price', 'stock', 'is_active',
            'category__name', 'created_by__username'
        )

    @action(detail=False, methods=['get'])
    def top_rated(self, request):
        """সবচেয়ে ভালো rated products"""
        products = self.get_queryset().filter(
            review_count__gte=5,
            avg_rating__gte=4.0
        ).order_by('-avg_rating')[:10]

        serializer = self.get_serializer(products, many=True)
        return Response(serializer.data)

# Redis ইনস্টল করুন
# Ubuntu/Debian:
sudo apt-get install redis-server

# Mac:
brew install redis

# Redis চালু করুন
redis-server

# Python Redis client ইনস্টল
pip install redis django-redis

# Cache Configuration
CACHES = {
    'default': {
        'BACKEND': 'django_redis.cache.RedisCache',
        'LOCATION': 'redis://127.0.0.1:6379/1',
        'OPTIONS': {
            'CLIENT_CLASS': 'django_redis.client.DefaultClient',
            'PARSER_CLASS': 'redis.connection.HiredisParser',
            'CONNECTION_POOL_KWARGS': {
                'max_connections': 50,
                'retry_on_timeout': True,
            },
            'SOCKET_CONNECT_TIMEOUT': 5,
            'SOCKET_TIMEOUT': 5,
        },
        'KEY_PREFIX': 'speedtest',
        'TIMEOUT': 300,  # 5 minutes default
    }
}

# Session cache
SESSION_ENGINE = 'django.contrib.sessions.backends.cache'
SESSION_CACHE_ALIAS = 'default'

from django.core.cache import cache
from django.utils.decorators import method_decorator
from django.views.decorators.cache import cache_page
from functools import wraps
import hashlib
import json

def cache_key_generator(prefix, *args, **kwargs):
    """
    Unique cache key তৈরি করুন
    """
    key_data = f"{prefix}:{args}:{sorted(kwargs.items())}"
    return hashlib.md5(key_data.encode()).hexdigest()


def smart_cache(timeout=300, key_prefix='view'):
    """
    Smart caching decorator যা query params consider করে
    """
    def decorator(func):
        @wraps(func)
        def wrapper(self, request, *args, **kwargs):
            # Cache key তৈরি করুন query params সহ
            cache_key = cache_key_generator(
                key_prefix,
                request.path,
                request.GET.dict(),
                request.user.id if request.user.is_authenticated else 'anonymous'
            )

            # Cache থেকে check করুন
            cached_response = cache.get(cache_key)
            if cached_response:
                print(f"✅ Cache HIT: {cache_key[:20]}...")
                return cached_response

            # Cache miss - function call করুন
            print(f"❌ Cache MISS: {cache_key[:20]}...")
            response = func(self, request, *args, **kwargs)

            # Response cache করুন
            cache.set(cache_key, response, timeout)
            return response

        return wrapper
    return decorator


def invalidate_product_cache(product_id=None):
    """
    Product cache invalidate করুন
    """
    if product_id:
        cache.delete(f'product:{product_id}')
    else:
        # সব product cache clear করুন
        cache.delete_pattern('product:*')

from rest_framework.views import APIView
from rest_framework.response import Response
from rest_framework.permissions import IsAuthenticated
from django.core.cache import cache
from django.utils.decorators import method_decorator
from django.views.decorators.cache import cache_page
from .models import Product
from .serializers import ProductSerializer
from .cache_utils import smart_cache

class CachedProductListView(APIView):
    """
    Redis cache সহ Product list
    """
    permission_classes = [IsAuthenticated]

    @smart_cache(timeout=300, key_prefix='product_list')
    def get(self, request):
        # Query params
        category = request.GET.get('category')
        min_price = request.GET.get('min_price')
        max_price = request.GET.get('max_price')

        # Build queryset
        queryset = Product.objects.select_related('category').filter(is_active=True)

        if category:
            queryset = queryset.filter(category__slug=category)
        if min_price:
            queryset = queryset.filter(price__gte=min_price)
        if max_price:
            queryset = queryset.filter(price__lte=max_price)

        # Serialize
        products = queryset[:100]
        serializer = ProductSerializer(products, many=True)

        return Response({
            'count': len(serializer.data),
            'results': serializer.data,
            'cached': False  # প্রথমবার
        })


class CachedProductDetailView(APIView):
    """
    Individual product cache
    """
    permission_classes = [IsAuthenticated]

    def get(self, request, pk):
        # Cache key
        cache_key = f'product:{pk}'

        # Check cache
        cached_data = cache.get(cache_key)
        if cached_data:
            return Response({
                'data': cached_data,
                'cached': True
            })

        # Fetch from DB
        try:
            product = Product.objects.select_related('category')\
                                    .prefetch_related('tags', 'reviews')\
                                    .get(pk=pk)
            serializer = ProductSerializer(product)

            # Cache করুন
            cache.set(cache_key, serializer.data, timeout=600)  # 10 minutes

            return Response({
                'data': serializer.data,
                'cached': False
            })
        except Product.DoesNotExist:
            return Response({'error': 'Not found'}, status=404)

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

@receiver(post_save, sender=Product)
def invalidate_product_cache_on_save(sender, instance, **kwargs):
    """
    Product save হলে cache clear করুন
    """
    cache_key = f'product:{instance.id}'
    cache.delete(cache_key)

    # List cache ও clear করুন
    cache.delete_pattern('product_list:*')

    print(f"🗑️ Cache cleared for product {instance.id}")


@receiver(post_delete, sender=Product)
def invalidate_product_cache_on_delete(sender, instance, **kwargs):
    """
    Product delete হলে cache clear করুন
    """
    cache_key = f'product:{instance.id}'
    cache.delete(cache_key)
    cache.delete_pattern('product_list:*')

    print(f"🗑️ Cache cleared for deleted product {instance.id}")

from django.http import JsonResponse
from django.views import View
from asgiref.sync import sync_to_async
from .models import Product
from .serializers import ProductSerializer
import asyncio

class AsyncProductListView(View):
    """
    Async view for better concurrency
    """

    async def get(self, request):
        # Async database query
        products = await sync_to_async(list)(
            Product.objects.select_related('category')
                          .filter(is_active=True)[:100]
        )

        # Async serialization
        serializer = ProductSerializer(products, many=True)
        data = await sync_to_async(lambda: serializer.data)()

        return JsonResponse({
            'count': len(data),
            'results': data
        })


class AsyncMultipleAPIView(View):
    """
    একসাথে multiple API call করুন
    """

    async def get(self, request):
        # Parallel queries
        products_task = sync_to_async(list)(
            Product.objects.filter(is_active=True)[:10]
        )

        categories_task = sync_to_async(list)(
            Category.objects.all()
        )

        # Wait for both
        products, categories = await asyncio.gather(
            products_task,
            categories_task
        )

        return JsonResponse({
            'products_count': len(products),
            'categories_count': len(categories)
        })

# Locust ইনস্টল করুন
pip install locust

from locust import HttpUser, task, between
import random

class ProductAPIUser(HttpUser):
    """
    Load testing for Product API
    """
    wait_time = between(1, 3)  # 1-3 seconds wait

    def on_start(self):
        """Login করুন"""
        response = self.client.post('/api-auth/login/', {
            'username': 'admin',
            'password': 'admin123'
        })
        if response.status_code == 200:
            print("✅ Login successful")

    @task(3)  # Weight: 3
    def list_products(self):
        """Product list API"""
        self.client.get('/api/products/')

    @task(2)  # Weight: 2
    def list_products_filtered(self):
        """Filtered product list"""
        min_price = random.randint(10, 100)
        self.client.get(f'/api/products/?min_price={min_price}')

    @task(1)  # Weight: 1
    def get_product_detail(self):
        """Single product detail"""
        product_id = random.randint(1, 1000)
        self.client.get(f'/api/products/{product_id}/')

    @task(3)
    def bolt_list_products(self):
        """Bolt optimized list"""
        self.client.get('/api/bolt/products/')

    @task(1)
    def cached_list_products(self):
        """Cached product list"""
        self.client.get('/api/cached/products/')


class HeavyLoadUser(HttpUser):
    """
    Heavy load testing
    """
    wait_time = between(0.5, 1)

    @task
    def rapid_requests(self):
        """দ্রুত requests"""
        endpoints = [
            '/api/products/',
            '/api/bolt/products/',
            '/api/cached/products/',
        ]
        endpoint = random.choice(endpoints)
        self.client.get(endpoint)

# Web UI সহ
locust -f locustfile.py --host=http://127.0.0.1:8000

# ব্রাউজারে খুলুন: http://localhost:8089

# অথবা headless mode
locust -f locustfile.py --host=http://127.0.0.1:8000 \
       --users 100 --spawn-rate 10 --run-time 1m --headless

pip install django-debug-toolbar

INSTALLED_APPS = [
    # ...
    'debug_toolbar',
]

MIDDLEWARE = [
    'debug_toolbar.middleware.DebugToolbarMiddleware',  # ✅ প্রথমে যোগ করুন
    # ... other middleware
]

INTERNAL_IPS = [
    '127.0.0.1',
]

# Debug Toolbar Configuration
DEBUG_TOOLBAR_CONFIG = {
    'SHOW_TOOLBAR_CALLBACK': lambda request: DEBUG,
}

from django.urls import path, include
from django.conf import settings

urlpatterns = [
    # ... your urls
]

if settings.DEBUG:
    import debug_toolbar
    urlpatterns = [
        path('__debug__/', include(debug_toolbar.urls)),
    ] + urlpatterns

LOGGING = {
    'version': 1,
    'disable_existing_loggers': False,
    'formatters': {
        'verbose': {
            'format': '{levelname} {asctime} {module} {message}',
            'style': '{',
        },
    },
    'handlers': {
        'file': {
            'level': 'INFO',
            'class': 'logging.FileHandler',
            'filename': 'django_debug.log',
            'formatter': 'verbose',
        },
        'console': {
            'level': 'DEBUG',
            'class': 'logging.StreamHandler',
            'formatter': 'verbose',
        },
    },
    'loggers': {
        'django': {
            'handlers': ['file', 'console'],
            'level': 'INFO',
            'propagate': True,
        },
        'products': {
            'handlers': ['file', 'console'],
            'level': 'DEBUG',
            'propagate': False,
        },
    },
}

import time
from django.db import connection
from django.utils.deprecation import MiddlewareMixin
import logging

logger = logging.getLogger('products')

class QueryCountDebugMiddleware(MiddlewareMixin):
    """
    প্রতিটি request এ query count দেখান
    """

    def process_request(self, request):
        request.start_time = time.time()
        request.query_count_start = len(connection.queries)

    def process_response(self, request, response):
        if hasattr(request, 'start_time'):
            # Calculate time
            total_time = time.time() - request.start_time

            # Calculate queries
            query_count = len(connection.queries) - request.query_count_start

            # Log
            logger.info(
                f"Path: {request.path} | "
                f"Time: {total_time:.2f}s | "
                f"Queries: {query_count}"
            )

            # Add to response header
            response['X-Query-Count'] = query_count
            response['X-Response-Time'] = f"{total_time:.2f}s"

        return response

MIDDLEWARE = [
    # ...
    'products.middleware.QueryCountDebugMiddleware',  # ✅ শেষে যোগ করুন
]

import requests
import time
import statistics
from requests.auth import HTTPBasicAuth
from tabulate import tabulate
import json

BASE_URL = 'http://127.0.0.1:8000/api'
USERNAME = 'admin'
PASSWORD = 'admin123'
auth = HTTPBasicAuth(USERNAME, PASSWORD)

def test_endpoint(url, name, iterations=20):
    """Detailed performance test"""
    print(f"\n🧪 Testing: {name}")

    times = []
    query_counts = []

    for i in range(iterations):
        start = time.time()
        response = requests.get(url, auth=auth)
        end = time.time()

        elapsed = (end - start) * 1000
        times.append(elapsed)

        # Get query count from header
        query_count = response.headers.get('X-Query-Count', 'N/A')
        if query_count != 'N/A':
            query_counts.append(int(query_count))

    # Statistics
    avg_time = statistics.mean(times)
    median_time = statistics.median(times)
    std_dev = statistics.stdev(times) if len(times) > 1 else 0
    min_time = min(times)
    max_time = max(times)

    avg_queries = statistics.mean(query_counts) if query_counts else 'N/A'

    return {
        'name': name,
        'avg': f"{avg_time:.2f}ms",
        'median': f"{median_time:.2f}ms",
        'min': f"{min_time:.2f}ms",
        'max': f"{max_time:.2f}ms",
        'std_dev': f"{std_dev:.2f}ms",
        'queries': f"{avg_queries:.1f}" if avg_queries != 'N/A' else 'N/A',
        'avg_raw': avg_time
    }

# Test করুন
print("=" * 80)
print("🚀 COMPREHENSIVE PERFORMANCE TEST")
print("=" * 80)

tests = [
    (f'{BASE_URL}/products/', 'Normal Django REST'),
    (f'{BASE_URL}/bolt/products/', 'Django Bolt'),
    (f'{BASE_URL}/cached/products/', 'Redis Cached'),
    (f'{BASE_URL}/products/?min_price=100', 'Normal (Filtered)'),
    (f'{BASE_URL}/bolt/products/?min_price=100', 'Bolt (Filtered)'),
]

results = []
for url, name in tests:
    result = test_endpoint(url, name, iterations=20)
    results.append(result)
    time.sleep(0.5)  # Cool down

# Display results
print("\n" + "=" * 80)
print("📊 RESULTS TABLE")
print("=" * 80)

table_data = [
    [r['name'], r['avg'], r['median'], r['min'], r['max'], r['std_dev'], r['queries']]
    for r in results
]

headers = ['Endpoint', 'Avg', 'Median', 'Min', 'Max', 'Std Dev', 'Queries']
print(tabulate(table_data, headers=headers, tablefmt='grid'))

# Speedup comparison
print("\n" + "=" * 80)
print("⚡ SPEEDUP COMPARISON")
print("=" * 80)

baseline = results[0]['avg_raw']
for result in results[1:]:
    speedup = baseline / result['avg_raw']
    print(f"{result['name']:30} → {speedup:.2f}x faster than baseline")

print("=" * 80)

pip install tabulate
python advanced_performance_test.py