⚙️ فصل ۱۰: مباحث پیشرفته

حالا که اصول کار می‌دونی، وقتشه با ابزارهای حرفه‌ای آشنا بشی!

📡 Signals – سیگنال‌ها

Signals مکانیزمی هستن که اجازه می‌دن وقتی یه اتفاق رخ داد، کارهای دیگه‌ای انجام بشه. مثل event در JS!

سیگنال‌های پرکاربرد

مثال ساده: ساخت Profile خودکار

from django.db.models.signals import post_save
from django.contrib.auth.models import User
from django.dispatch import receiver
from .models import Profile

@receiver(post_save, sender=User)
def create_profile(sender, instance, created, **kwargs):
    """وقتی کاربر ساخته شد، پروفایل هم ساخته بشه"""
    if created:
        Profile.objects.create(user=instance)

@receiver(post_save, sender=User)
def save_profile(sender, instance, **kwargs):
    """با ذخیره کاربر، پروفایل هم ذخیره بشه"""
    instance.profile.save()

فعال‌سازی سیگنال‌ها

from django.apps import AppConfig

class AccountsConfig(AppConfig):
    default_auto_field = 'django.db.models.BigAutoField'
    name = 'accounts'
    
    def ready(self):
        # وقتی اپ آماده شد، سیگنال‌ها رو ایمپورت کن
        import accounts.signals

مثال‌های دیگه

from django.db.models.signals import pre_save, post_delete
from django.dispatch import receiver
from django.utils.text import slugify
from .models import Product

@receiver(pre_save, sender=Product)
def auto_slug(sender, instance, **kwargs):
    """قبل از ذخیره، slug رو خودکار بساز"""
    if not instance.slug:
        instance.slug = slugify(instance.name, allow_unicode=True)

@receiver(post_delete, sender=Product)
def delete_image_file(sender, instance, **kwargs):
    """بعد از حذف محصول، تصویرش هم حذف بشه"""
    if instance.image:
        instance.image.delete(save=False)

🔄 Middleware

Middleware لایه‌ای بین درخواست و پاسخ هست که می‌تونه قبل/بعد از پردازش هر درخواست کاری انجام بده.

ساخت Middleware سفارشی

import time

class TimingMiddleware:
    def __init__(self, get_response):
        self.get_response = get_response
    
    def __call__(self, request):
        # کد قبل از پردازش درخواست
        start_time = time.time()
        
        response = self.get_response(request)
        
        # کد بعد از پردازش
        duration = time.time() - start_time
        response['X-Response-Time'] = f'{duration:.2f}s'
        print(f'درخواست {request.path} در {duration:.2f} ثانیه پردازش شد')
        
        return response

ثبت Middleware

MIDDLEWARE = [
    'django.middleware.security.SecurityMiddleware',
    'django.contrib.sessions.middleware.SessionMiddleware',
    'django.middleware.common.CommonMiddleware',
    'django.middleware.csrf.CsrfViewMiddleware',
    'django.contrib.auth.middleware.AuthenticationMiddleware',
    'django.contrib.messages.middleware.MessageMiddleware',
    'django.middleware.clickjacking.XFrameOptionsMiddleware',
    
    # Middleware سفارشی
    'myapp.middleware.TimingMiddleware',
]

مثال: Middleware برای IP بلوک کردن

from django.http import HttpResponseForbidden

class BlockIPMiddleware:
    BLOCKED_IPS = ['1.2.3.4', '5.6.7.8']
    
    def __init__(self, get_response):
        self.get_response = get_response
    
    def __call__(self, request):
        # دریافت IP کاربر
        ip = self.get_client_ip(request)
        
        if ip in self.BLOCKED_IPS:
            return HttpResponseForbidden('دسترسی شما مسدود است')
        
        return self.get_response(request)
    
    def get_client_ip(self, request):
        x_forwarded = request.META.get('HTTP_X_FORWARDED_FOR')
        if x_forwarded:
            return x_forwarded.split(',')[0]
        return request.META.get('REMOTE_ADDR')

💾 Caching – کش

کش کردن نتایج گران‌قیمت، سرعت سایتت رو خیلی بالا می‌بره.

تنظیمات کش

# کش حافظه (ساده، برای توسعه)
CACHES = {
    'default': {
        'BACKEND': 'django.core.cache.backends.locmem.LocMemCache',
        'LOCATION': 'unique-snowflake',
    }
}

# کش با Redis (برای production)
CACHES = {
    'default': {
        'BACKEND': 'django.core.cache.backends.redis.RedisCache',
        'LOCATION': 'redis://127.0.0.1:6379/1',
    }
}

# کش با فایل
CACHES = {
    'default': {
        'BACKEND': 'django.core.cache.backends.filebased.FileBasedCache',
        'LOCATION': '/var/tmp/django_cache',
    }
}

روش‌های استفاده از کش

۱. کش کردن کل صفحه

from django.views.decorators.cache import cache_page

@cache_page(60 * 15)  # ۱۵ دقیقه کش
def product_list(request):
    products = Product.objects.all()
    return render(request, 'list.html', {'products': products})

۲. کش کردن دستی

from django.core.cache import cache

def expensive_view(request):
    # خواندن از کش
    data = cache.get('expensive_data')
    
    if data is None:
        # محاسبه گران‌قیمت
        data = calculate_expensive_data()
        # ذخیره در کش برای ۱ ساعت
        cache.set('expensive_data', data, 3600)
    
    return JsonResponse(data)

# حذف کش
cache.delete('expensive_data')

# پاک کردن کل کش
cache.clear()

۳. کش در قالب

{% load cache %}

{# کش این قسمت برای ۵ دقیقه #}
{% cache 300 sidebar request.user.username %}
    <div class="sidebar">
        ... محتوای ثابت ...
    </div>
{% endcache %}

🌿 Celery – کارهای پس‌زمینه

برای کارهای زمان‌بر (ارسال ایمیل، پردازش فایل) که نباید کاربر منتظر بمونه، از Celery استفاده می‌کنیم.

نصب

pip install celery redis

تنظیمات

import os
from celery import Celery

os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'myproject.settings')

app = Celery('myproject')
app.config_from_object('django.conf:settings', namespace='CELERY')
app.autodiscover_tasks()

from .celery import app as celery_app
__all__ = ('celery_app',)

CELERY_BROKER_URL = 'redis://localhost:6379/0'
CELERY_RESULT_BACKEND = 'redis://localhost:6379/0'
CELERY_TIMEZONE = 'Asia/Tehran'

تعریف Task

from celery import shared_task
from django.core.mail import send_mail

@shared_task
def send_welcome_email(user_email, username):
    send_mail(
        subject='به ICSD خوش آمدید',
        message=f'سلام {username}، حسابت با موفقیت ساخته شد!',
        from_email='noreply@icsd.ir',
        recipient_list=[user_email],
    )

@shared_task
def process_image(image_path):
    """پردازش تصویر در پس‌زمینه"""
    from PIL import Image
    img = Image.open(image_path)
    img.thumbnail((800, 800))
    img.save(image_path, optimize=True)

اجرای Task

from .tasks import send_welcome_email

def register(request):
    # ... ثبت‌نام ...
    
    # ارسال ایمیل در پس‌زمینه (کاربر منتظر نمی‌مونه!)
    send_welcome_email.delay(user.email, user.username)
    
    return redirect('home')

اجرای Worker

celery -A myproject worker -l info

کارهای زمان‌بندی‌شده

from celery.schedules import crontab

CELERY_BEAT_SCHEDULE = {
    'cleanup-old-logs': {
        'task': 'myapp.tasks.cleanup_logs',
        'schedule': crontab(hour=3, minute=0),  # هر روز ساعت ۳ صبح
    },
    'send-daily-report': {
        'task': 'reports.tasks.daily_report',
        'schedule': crontab(hour=8, minute=0, day_of_week=1),  # دوشنبه ۸ صبح
    },
}

🛠️ Custom Management Commands

برای ساخت دستورات سفارشی python manage.py ...

products/
├── management/
│   ├── __init__.py
│   └── commands/
│       ├── __init__.py
│       └── update_prices.py
├── models.py
└── ...

from django.core.management.base import BaseCommand
from products.models import Product

class Command(BaseCommand):
    help = 'به‌روزرسانی قیمت محصولات'
    
    def add_arguments(self, parser):
        parser.add_argument('percent', type=int, help='درصد افزایش')
        parser.add_argument(
            '--category',
            type=str,
            help='فقط یک دسته خاص'
        )
    
    def handle(self, *args, **options):
        percent = options['percent']
        category = options.get('category')
        
        products = Product.objects.all()
        if category:
            products = products.filter(category__slug=category)
        
        count = 0
        for product in products:
            product.price = product.price * (1 + percent / 100)
            product.save()
            count += 1
        
        self.stdout.write(self.style.SUCCESS(
            f'✅ {count} محصول {percent}٪ افزایش قیمت یافت'
        ))

اجرای دستور

python manage.py update_prices 10
python manage.py update_prices 15 --category=carpets

🎯 مثال‌های کاربردی

class AuditLog(models.Model):
    user = models.ForeignKey(User, on_delete=models.SET_NULL, null=True)
    action = models.CharField(max_length=50)
    model = models.CharField(max_length=100)
    object_id = models.IntegerField()
    timestamp = models.DateTimeField(auto_now_add=True)
    ip_address = models.GenericIPAddressField(null=True)

from django.db.models.signals import post_save, post_delete
from django.dispatch import receiver
from products.models import Product
from .models import AuditLog

@receiver(post_save, sender=Product)
def log_product_save(sender, instance, created, **kwargs):
    AuditLog.objects.create(
        action='create' if created else 'update',
        model='Product',
        object_id=instance.id
    )

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

def get_featured_products():
    """گرفتن محصولات ویژه با کش"""
    cached = cache.get('featured_products')
    if cached is not None:
        return cached
    
    products = list(Product.objects.filter(
        is_featured=True,
        is_active=True
    ).select_related('category')[:10])
    
    cache.set('featured_products', products, 3600)  # ۱ ساعت
    return products

@receiver([post_save, post_delete], sender=Product)
def invalidate_featured_cache(sender, instance, **kwargs):
    """با هر تغییر محصول، کش پاک شه"""
    cache.delete('featured_products')

📝 تمرین‌ها