Django Signals are a mechanism for decoupled apps to communicate when certain events occur (model save, delete, user login, etc.). They follow the Observer pattern without direct coupling between sender and receiver.

from django.db.models.signals import post_save, post_delete
from django.dispatch import receiver
from .models import Book

# Method 1: Using @receiver decorator
@receiver(post_save, sender=Book)
def send_email_on_book_created(sender, instance, created, **kwargs):
    if created:
        print(f"New book created: {instance.title}")
        send_notification_email(instance)

@receiver(post_delete, sender=Book)
def log_book_deletion(sender, instance, **kwargs):
    print(f"Book deleted: {instance.title}")

# Method 2: Manual signal connection
from django.db.models.signals import post_save

def update_book_cache(sender, instance, created, **kwargs):
    if created:
        cache.set(f"book_{instance.id}", instance, timeout=3600)

post_save.connect(update_book_cache, sender=Book)

# Method 3: In apps.py (Recommended)
from django.apps import AppConfig
from django.db.models.signals import post_save

class MyAppConfig(AppConfig):
    default_auto_field = "django.db.models.BigAutoField"
    name = "myapp"
    
    def ready(self):
        # Signals imported here to avoid circular imports
        from .signals import send_email_on_book_created
        post_save.connect(send_email_on_book_created, sender=self.apps.get_model("myapp", "Book"))

# Best Practice: signals.py file
# myapp/signals.py
from django.db.models.signals import post_save
from django.dispatch import receiver
from .models import Book

@receiver(post_save, sender=Book)
def update_related_data(sender, instance, created, **kwargs):
    if created:
        # Create related records
        Category.objects.get_or_create(name=instance.category)

# myapp/apps.py
from django.apps import AppConfig

class MyAppConfig(AppConfig):
    name = "myapp"
    
    def ready(self):
        import myapp.signals  # Import signals when app is ready

Decorators are functions that modify or enhance other functions/classes without changing their source code. They wrap functions to add new functionality like authentication checks, logging, caching, or timing.

// Basic Python Decorator
def my_decorator(func):
    def wrapper(*args, **kwargs):
        print(f"Before calling {func.__name__}")
        result = func(*args, **kwargs)
        print(f"After calling {func.__name__}")
        return result
    return wrapper

@my_decorator
def greet(name):
    print(f"Hello {name}")

greet("John")  # Prints: Before, Hello John, After

// Django View Decorator - Authentication
from django.contrib.auth.decorators import login_required
from django.http import HttpResponse

@login_required(login_url="/login/")
def dashboard(request):
    return HttpResponse("Welcome to dashboard")
    # Redirects to /login/ if not authenticated

// Custom Decorator - Timing
import time
from functools import wraps

def timing_decorator(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        start = time.time()
        result = func(*args, **kwargs)
        end = time.time()
        print(f"{func.__name__} took {end-start:.2f} seconds")
        return result
    return wrapper

@timing_decorator
def slow_function():
    time.sleep(2)
    return "Done"

// Custom Decorator - Logging
from functools import wraps
import logging

logger = logging.getLogger(__name__)

def log_request(func):
    @wraps(func)
    def wrapper(request, *args, **kwargs):
        logger.info(f"User {request.user} accessing {func.__name__}")
        try:
            response = func(request, *args, **kwargs)
            logger.info(f"Response status: {response.status_code}")
            return response
        except Exception as e:
            logger.error(f"Error in {func.__name__}: {str(e)}")
            raise
    return wrapper

@log_request
def view_books(request):
    return JsonResponse({"books": [...]})

// Custom Decorator with Parameters
def rate_limit(calls_per_minute=60):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            # Implement rate limiting logic
            return func(*args, **kwargs)
        return wrapper
    return decorator

@rate_limit(calls_per_minute=100)
def api_endpoint(request):
    return JsonResponse({"data": []})

// Class-Based View Decorator
from django.utils.decorators import method_decorator
from django.views import View

@method_decorator(login_required, name="dispatch")
class BookListView(View):
    def get(self, request):
        return JsonResponse({"books": []})

// Multiple Decorators
from django.contrib.auth.decorators import login_required, permission_required

@login_required
@permission_required("myapp.can_edit_books")
def edit_book(request, book_id):
    return JsonResponse({"status": "editing"})

Signals are powerful but can cause confusion and performance issues if misused. Follow best practices to avoid circular imports, performance problems, and debugging difficulties.

// ✅ CORRECT - signals.py with apps.py
// myapp/signals.py
from django.db.models.signals import post_save
from django.dispatch import receiver
from .models import Book

@receiver(post_save, sender=Book)
def update_cache(sender, instance, created, **kwargs):
    # Keep this lightweight
    cache.set(f"book_{instance.id}", instance)

// myapp/apps.py
from django.apps import AppConfig

class MyAppConfig(AppConfig):
    name = "myapp"
    
    def ready(self):
        import myapp.signals  # Import signals

// ❌ WRONG - Signal in models.py (circular import risk)
// models.py - DON't do this!
from django.db.models.signals import post_save

@receiver(post_save, sender=Book)
def update_cache(sender, instance, **kwargs):
    pass  # Causes circular import issues

// Using dispatch_uid
@receiver(post_save, sender=Book, dispatch_uid="update_book_cache")
def update_cache(sender, instance, **kwargs):
    # dispatch_uid prevents duplicate connections
    pass

// ✅ Best: Async heavy operations
from celery import shared_task

@receiver(post_save, sender=Book)
def trigger_email(sender, instance, created, **kwargs):
    if created:
        send_email_async.delay(instance.id)

@shared_task
def send_email_async(book_id):
    book = Book.objects.get(id=book_id)
    send_notification_email(book)

// ❌ AVOID: Heavy queries in signals
@receiver(post_save, sender=Book)
def bad_handler(sender, instance, **kwargs):
    # Bad: This will be called for EVERY save
    all_books = Book.objects.all()  # Heavy query
    for book in all_books:
        book.update_something()

// ✅ CORRECT: Use select_related to optimize
@receiver(post_save, sender=Book)
def good_handler(sender, instance, **kwargs):
    # Only update related object
    instance.category.update_book_count()

// ✅ CORRECT: Disconnect signals in tests
from django.db.models.signals import post_save
from django.test import TestCase

class BookTestCase(TestCase):
    @classmethod
    def setUpClass(cls):
        super().setUpClass()
        post_save.disconnect(update_cache, sender=Book)
    
    @classmethod
    def tearDownClass(cls):
        super().tearDownClass()
        post_save.connect(update_cache, sender=Book)

Context managers handle resource allocation and cleanup (files, database connections). They use `__enter__` and `__exit__` methods or @contextmanager decorator for cleaner syntax.

// Context Manager - Class Based
class DatabaseConnection:
    def __enter__(self):
        print("Opening database connection")
        self.db = connect_to_db()
        return self.db
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        print("Closing database connection")
        self.db.close()
        return False  # Don't suppress exceptions

// Usage with 'with' statement
with DatabaseConnection() as db:
    result = db.query("SELECT * FROM books")
    print(result)
# Connection automatically closed

// Context Manager Decorator (@contextmanager)
from contextlib import contextmanager

@contextmanager
def database_connection():
    db = connect_to_db()
    try:
        yield db
    finally:
        db.close()

// Usage
with database_connection() as db:
    result = db.query("SELECT * FROM books")

// Practical Example - File Handling
# Python handles this automatically
with open("file.txt", "r") as f:
    content = f.read()
# File closed automatically

// Custom Context Manager - Transaction
from contextlib import contextmanager

@contextmanager
def atomic_transaction(db):
    try:
        db.begin()
        yield db
        db.commit()
    except Exception as e:
        db.rollback()
        raise

// Usage
with atomic_transaction(db) as txn:
    txn.execute("UPDATE books SET price = 100")
    txn.execute("UPDATE inventory SET count = 50")

// Django's Context Manager
from django.db import transaction

with transaction.atomic():
    book = Book.objects.create(title="New Book")
    review = Review.objects.create(book=book, rating=5)
    # Both operations atomic - rollback if either fails

// Suppress Exceptions
from contextlib import suppress

with suppress(FileNotFoundError):
    os.remove("file.txt")  # Won't raise if file doesn't exist

// Redirect Context
from contextlib import redirect_stdout
import io

output = io.StringIO()
with redirect_stdout(output):
    print("Hello World")

result = output.getvalue()  # "Hello World\
"

Signal handlers are executed synchronously in the same process, blocking the request. For long-running operations, use Celery tasks to execute asynchronously and avoid blocking.

// ❌ SYNCHRONOUS - Blocks request
@receiver(post_save, sender=Book)
def send_email_on_save(sender, instance, created, **kwargs):
    if created:
        # This BLOCKS the request until email is sent
        send_notification_email(instance.author_email)
        print("Email sent")  # Delays user response

// ✅ ASYNCHRONOUS - Celery Queue
from celery import shared_task
from django.db.models.signals import post_save

@receiver(post_save, sender=Book)
def trigger_email_task(sender, instance, created, **kwargs):
    if created:
        # Queue task for async execution
        send_email_task.delay(instance.id)

@shared_task
def send_email_task(book_id):
    book = Book.objects.get(id=book_id)
    send_notification_email(book.author_email)

// ✅ SEMI-SYNC - Threading (light workload)
import threading

@receiver(post_save, sender=Book)
def async_logging(sender, instance, created, **kwargs):
    def log_in_background():
        AuditLog.objects.create(
            action="created" if created else "updated",
            model="Book",
            instance_id=instance.id
        )
    
    thread = threading.Thread(target=log_in_background)
    thread.daemon = True
    thread.start()

// Signal Handler Timing Example
import time

@receiver(post_save, sender=Book)
def slow_handler(sender, instance, created, **kwargs):
    start = time.time()
    time.sleep(5)  # Simulates slow operation
    duration = time.time() - start
    print(f"Handler took {duration} seconds - USER WAITED THIS LONG!")

// Measure Impact
from django.test import TestCase
import time

class SignalPerformanceTest(TestCase):
    def test_save_speed(self):
        start = time.time()
        Book.objects.create(title="Test", author="Test")
        duration = time.time() - start
        print(f"Save took {duration} seconds")
        # With async: ~0.01 seconds
        # With sync email: ~2 seconds (blocked)

// Best Practice - Use Celery for Heavy Operations
from celery import shared_task

@receiver(post_save, sender=Book)
def handle_book_creation(sender, instance, created, **kwargs):
    if created:
        # Quick operations in signal
        cache.set(f"book_{instance.id}", instance)
        
        # Heavy operations to Celery
        process_book_async.delay(instance.id)

@shared_task
def process_book_async(book_id):
    book = Book.objects.get(id=book_id)
    # Send notifications
    send_email_task(book)
    # Process images
    generate_thumbnails(book)
    # Update search index
    update_search_index(book)

Multiple decorators are applied bottom-up during definition but executed top-down during function call. Understanding the order is critical to avoid bugs.

// Simple Decorator Chain
def decorator1(func):
    print("D1 decorated")
    def wrapper(*args, **kwargs):
        print("D1 before")
        result = func(*args, **kwargs)
        print("D1 after")
        return result
    return wrapper

def decorator2(func):
    print("D2 decorated")
    def wrapper(*args, **kwargs):
        print("D2 before")
        result = func(*args, **kwargs)
        print("D2 after")
        return result
    return wrapper

// Order of execution
@decorator1
@decorator2
def my_function():
    print("Function executed")

my_function()

// EXECUTION OUTPUT:
// D2 decorated (bottom decorator applied first)
// D1 decorated (top decorator applied second)
// D1 before (top decorator executes first)
// D2 before
// Function executed
// D2 after
// D1 after (top decorator executes last)

// Django Example - Multiple Permissions
from django.contrib.auth.decorators import login_required, permission_required

@login_required
@permission_required("myapp.can_edit_books")
def edit_book(request, book_id):
    return JsonResponse({"status": "editing"})

// EXECUTION ORDER:
// 1. Check if user is authenticated (login_required)
// 2. Check if user has permission (permission_required)
// 3. Execute edit_book view

// Practical Django Example
def timing_decorator(func):
    def wrapper(*args, **kwargs):
        import time
        start = time.time()
        result = func(*args, **kwargs)
        print(f"Execution time: {time.time() - start}s")
        return result
    return wrapper

def auth_decorator(func):
    def wrapper(request, *args, **kwargs):
        if not request.user.is_authenticated:
            return redirect("login")
        return func(request, *args, **kwargs)
    return wrapper

@timing_decorator
@auth_decorator
def dashboard(request):
    return render(request, "dashboard.html")

// EXECUTION ORDER on call:
// 1. auth_decorator checks authentication
// 2. dashboard executes
// 3. timing_decorator measures time

// Complex Chain with Caching
from functools import wraps

def cache_result(timeout=300):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            cache_key = f"{func.__name__}_{str(args)}_{str(kwargs)}"
            result = cache.get(cache_key)
            if result is None:
                result = func(*args, **kwargs)
                cache.set(cache_key, result, timeout)
            return result
        return wrapper
    return decorator

def log_execution(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        logger.info(f"Calling {func.__name__}")
        return func(*args, **kwargs)
    return wrapper

@log_execution
@cache_result(timeout=600)
def expensive_operation(n):
    time.sleep(2)
    return n * 2

// EXECUTION:
// 1. log_execution wrapper called
// 2. cache_result wrapper checks cache
// 3. expensive_operation executes (if not cached)
// 4. result cached and returned

Custom view decorators wrap view functions to add cross-cutting concerns like authentication, request logging, rate limiting, and response modification without duplicating code.

// Custom Auth Decorator
from functools import wraps
from django.http import JsonResponse
from django.contrib.auth.decorators import login_required

def admin_only(func):
    @wraps(func)
    def wrapper(request, *args, **kwargs):
        if not request.user.is_staff:
            return JsonResponse({"error": "Admin access required"}, status=403)
        return func(request, *args, **kwargs)
    return wrapper

@admin_only
def delete_user(request, user_id):
    User.objects.get(id=user_id).delete()
    return JsonResponse({"status": "deleted"})

// Custom Logging Decorator
import logging
from functools import wraps

logger = logging.getLogger(__name__)

def log_request(func):
    @wraps(func)
    def wrapper(request, *args, **kwargs):
        logger.info(f"User {request.user} accessing {func.__name__}")
        logger.info(f"Method: {request.method}, Path: {request.path}")
        
        try:
            response = func(request, *args, **kwargs)
            logger.info(f"Response status: {response.status_code}")
            return response
        except Exception as e:
            logger.error(f"Error in {func.__name__}: {str(e)}", exc_info=True)
            raise
    return wrapper

@log_request
def dashboard(request):
    return render(request, "dashboard.html")

// Rate Limiting Decorator
from functools import wraps
from django.core.cache import cache
from django.http import JsonResponse

def rate_limit(calls_per_minute=60):
    def decorator(func):
        @wraps(func)
        def wrapper(request, *args, **kwargs):
            cache_key = f"ratelimit_{request.user.id}_{func.__name__}"
            request_count = cache.get(cache_key, 0)
            
            if request_count >= calls_per_minute:
                return JsonResponse(
                    {"error": "Rate limit exceeded"},
                    status=429
                )
            
            cache.set(cache_key, request_count + 1, 60)  # 60 seconds
            return func(request, *args, **kwargs)
        return wrapper
    return decorator

@rate_limit(calls_per_minute=100)
def api_endpoint(request):
    return JsonResponse({"data": [...]})

// Combined Decorators
from django.contrib.auth.decorators import login_required

@login_required
@rate_limit(calls_per_minute=200)
@log_request
def protected_api(request):
    return JsonResponse({"data": "sensitive"})

// EXECUTION ORDER:
// 1. Check authentication
// 2. Check rate limit
// 3. Log request
// 4. Execute view

// Decorator with Arguments
def require_role(role):
    def decorator(func):
        @wraps(func)
        def wrapper(request, *args, **kwargs):
            if not hasattr(request.user, "profile") or request.user.profile.role != role:
                return JsonResponse({"error": "Insufficient permissions"}, status=403)
            return func(request, *args, **kwargs)
        return wrapper
    return decorator

@require_role("editor")
def publish_article(request, article_id):
    article = Article.objects.get(id=article_id)
    article.published = True
    article.save()
    return JsonResponse({"status": "published"})

// Deprecation Warning Decorator
import warnings
from functools import wraps

def deprecated(replacement=None):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            msg = f"{func.__name__} is deprecated"
            if replacement:
                msg += f". Use {replacement} instead"
            warnings.warn(msg, DeprecationWarning, stacklevel=2)
            return func(*args, **kwargs)
        return wrapper
    return decorator

@deprecated(replacement="new_api_endpoint")
def old_api_endpoint(request):
    return JsonResponse({"data": []})

Test signals by disconnecting them to isolate behavior, using mock objects to verify calls, and checking that expected side effects occur. Signals can be tricky to test without proper isolation.

// Testing Signal Handlers
from django.test import TestCase
from django.db.models.signals import post_save, post_delete
from unittest.mock import patch, MagicMock
from .models import Book
from .signals import send_email_on_book_created

class SignalTestCase(TestCase):
    def setUp(self):
        # Disconnect signals for cleaner tests
        post_save.disconnect(send_email_on_book_created, sender=Book)
    
    def tearDown(self):
        # Reconnect signals after test
        post_save.connect(send_email_on_book_created, sender=Book)
    
    // Test signal is called
    @patch("myapp.signals.send_notification_email")
    def test_email_sent_on_book_creation(self, mock_email):
        book = Book.objects.create(
            title="Test Book",
            author="Test Author",
            price=9.99
        )
        
        # Manually trigger signal for isolated test
        send_email_on_book_created(sender=Book, instance=book, created=True)
        
        # Verify email was called
        mock_email.assert_called_once_with(book)
    
    // Test signal with mock
    @patch("myapp.signals.update_cache")
    def test_cache_updated(self, mock_cache):
        book = Book.objects.create(title="Test", author="Author", price=10)
        
        mock_cache.assert_called_once()
        call_args = mock_cache.call_args
        self.assertEqual(call_args[0][1].title, "Test")

// Test with side effects
class SignalSideEffectTest(TestCase):
    def test_book_update_creates_audit_log(self):
        book = Book.objects.create(title="Original", author="Author", price=10)
        
        # Update book
        book.title = "Updated"
        book.save()
        
        # Check if audit log was created
        from .models import AuditLog
        audit = AuditLog.objects.get(object_id=book.id)
        self.assertEqual(audit.action, "updated")
        self.assertEqual(audit.old_value, "Original")
        self.assertEqual(audit.new_value, "Updated")

// Test signal not called
class SignalNotCalledTest(TestCase):
    @patch("myapp.signals.expensive_operation")
    def test_expensive_operation_not_called_on_update(self, mock_op):
        book = Book.objects.create(title="Test", author="Author", price=10)
        mock_op.reset_mock()
        
        # Update shouldn't trigger expensive operation
        book.price = 15
        book.save()
        
        mock_op.assert_not_called()

// Integration Test - with real side effects
class SignalIntegrationTest(TestCase):
    def test_creating_book_updates_category_count(self):
        category = Category.objects.create(name="Fiction")
        initial_count = category.book_count
        
        book = Book.objects.create(
            title="Test",
            author="Author",
            price=10,
            category=category
        )
        
        category.refresh_from_db()
        self.assertEqual(category.book_count, initial_count + 1)

// Disabling signals in bulk operations
class BulkOperationTest(TestCase):
    @patch("django.db.models.signals.post_save.send")
    def test_bulk_create_efficiency(self, mock_signal):
        # Bulk operations bypass signals for efficiency
        books = [
            Book(title=f"Book {i}", author="Author", price=10)
            for i in range(100)
        ]
        
        created = Book.objects.bulk_create(books)
        
        # Signals are NOT sent for bulk_create
        self.assertEqual(len(created), 100)
        mock_signal.assert_not_called()

Both decorators and signals add overhead. Minimize expensive operations in decorators, use caching strategically, and profile your code to identify bottlenecks. Async operations prevent request blocking.

// ❌ SLOW - Heavy decorator
def check_permission_decorator(func):
    def wrapper(request, *args, **kwargs):
        # This runs on EVERY request
        all_permissions = Permission.objects.all()  # Heavy query
        allowed_perms = set(p.id for p in all_permissions)
        
        if request.user.id not in allowed_perms:
            return JsonResponse({"error": "No access"}, status=403)
        return func(request, *args, **kwargs)
    return wrapper

// ✅ FAST - Cached decorator
from django.core.cache import cache

def check_permission_decorator_cached(func):
    def wrapper(request, *args, **kwargs):
        cache_key = f"perms_{request.user.id}"
        permissions = cache.get(cache_key)
        
        if permissions is None:
            permissions = set(
                request.user.groups.values_list("permissions__id", flat=True)
            )
            cache.set(cache_key, permissions, 3600)
        
        if not permissions:
            return JsonResponse({"error": "No access"}, status=403)
        return func(request, *args, **kwargs)
    return wrapper

// Performance Testing
import time
from django.test import TestCase

class DecoratorPerformanceTest(TestCase):
    def test_decorator_overhead(self):
        def slow_decorator(func):
            def wrapper(*args, **kwargs):
                time.sleep(0.1)  # Simulates 100ms overhead
                return func(*args, **kwargs)
            return wrapper
        
        @slow_decorator
        def view(request):
            return JsonResponse({"data": "response"})
        
        start = time.time()
        for _ in range(100):
            view(MagicMock())
        total = time.time() - start
        
        print(f"100 requests: {total}s (10s overhead from decorator)")

// ❌ SLOW - Signal with N+1 queries
@receiver(post_save, sender=Book)
def update_categories(sender, instance, **kwargs):
    # N+1 problem: queries all related categories
    for category in Category.objects.all():
        category.book_count = category.books.count()
        category.save()

// ✅ FAST - Optimized signal
@receiver(post_save, sender=Book)
def update_category_optimized(sender, instance, **kwargs):
    # Only update related category
    if instance.category:
        instance.category.book_count = instance.category.books.count()
        instance.category.save()

// Profile Decorators
from django.test.utils import override_settings
import cProfile
import pstats

@override_settings(DEBUG=True)
def profile_view():
    pr = cProfile.Profile()
    pr.enable()
    
    # Run expensive view
    expensive_view(MagicMock())
    
    pr.disable()
    ps = pstats.Stats(pr)
    ps.print_stats()  # Shows where time is spent

// Minimize Decorator Stack
# ❌ Multiple decorators per view
@decorator1
@decorator2
@decorator3
@decorator4
@expensive_decorator
def view1(request):
    pass

# ✅ Combine related decorators
def combined_auth_and_logging(func):
    def wrapper(request, *args, **kwargs):
        # Do both auth and logging efficiently
        logger.info(f"Request from {request.user}")
        if not request.user.is_authenticated:
            return redirect("login")
        return func(request, *args, **kwargs)
    return wrapper

@combined_auth_and_logging
def view2(request):
    pass

Advanced signal patterns involve chaining signals (one signal triggers another), cascading effects through related models, and coordinating multiple signal handlers for complex workflows.

// Signal Cascading - One signal triggers another
from django.db.models.signals import post_save, post_delete

@receiver(post_save, sender=Book)
def update_author_stats(sender, instance, created, **kwargs):
    # When book is created/updated, update author stats
    author = instance.author
    author.total_books = author.books.count()
    author.avg_rating = author.books.aggregate(Avg("rating"))["rating__avg"]
    author.save()  # This triggers post_save for Author

@receiver(post_save, sender=Author)
def update_publisher_stats(sender, instance, **kwargs):
    # When author is saved (by signal above), update publisher stats
    publisher = instance.publisher
    publisher.total_authors = publisher.authors.count()
    publisher.save()  # Cascading continues

// Signal Coordination - Multiple handlers for same event
@receiver(post_save, sender=Book)
def handler1_cache_invalidation(sender, instance, **kwargs):
    cache.delete(f"books_list")

@receiver(post_save, sender=Book)
def handler2_search_index_update(sender, instance, **kwargs):
    update_search_index(instance)

@receiver(post_save, sender=Book)
def handler3_send_notification(sender, instance, **kwargs):
    notify_followers.delay(instance.id)

// ❌ PROBLEM - Circular Signal References
@receiver(post_save, sender=Book)
def update_category(sender, instance, **kwargs):
    category = instance.category
    category.book_count += 1
    category.save()  # Triggers post_save for Category

@receiver(post_save, sender=Category)
def update_books(sender, instance, **kwargs):
    # This could trigger book signal again!
    for book in instance.books.all():
        book.category_updated = True
        book.save()  # Back to book signal - circular!

// ✅ SOLUTION - Flag to prevent recursion
@receiver(post_save, sender=Book)
def update_category_safe(sender, instance, **kwargs):
    # Check if we're already in signal chain
    if getattr(instance, "_updating_category", False):
        return
    
    try:
        instance._updating_category = True
        category = instance.category
        category.book_count = category.books.count()
        category.save()
    finally:
        instance._updating_category = False

// Signal Chain Tracking - Debug cascading signals
class SignalLog:
    chain = []
    
    @classmethod
    def log_signal(cls, signal_name, model_name, action):
        cls.chain.append({
            "signal": signal_name,
            "model": model_name,
            "action": action,
            "time": timezone.now()
        })

@receiver(post_save, sender=Book)
def track_book_signal(sender, instance, created, **kwargs):
    SignalLog.log_signal("post_save", "Book", "created" if created else "updated")
    # Process...

@receiver(post_save, sender=Author)
def track_author_signal(sender, instance, **kwargs):
    SignalLog.log_signal("post_save", "Author", "updated")
    # Process...

// Async Signal Chain
from celery import chain, group

@receiver(post_save, sender=Book)
def process_book_async_chain(sender, instance, created, **kwargs):
    if created:
        # Chain: Task1 -> Task2 -> Task3
        workflow = chain(
            validate_book.s(instance.id),
            generate_thumbnail.s(),
            update_search_index.s()
        )
        workflow.apply_async()

// Signal Priority/Ordering
# Signals don't have built-in ordering, so use handler naming
# or explicit dispatcher management

def trigger_high_priority_signal():
    # Execute handlers in specific order
    handlers = [
        (handle_cache, 1),      # Priority 1 (highest)
        (handle_db_update, 2),  # Priority 2
        (handle_notification, 3) # Priority 3 (lowest)
    ]
    
    for handler, _ in sorted(handlers, key=lambda x: x[1]):
        handler()

// Conditional Signal Chaining
@receiver(post_save, sender=Order)
def process_order(sender, instance, created, **kwargs):
    if created:
        # Only chain if order is valid
        if instance.is_valid():
            order_chain = chain(
                process_payment.s(instance.id),
                send_confirmation.s(),
                update_inventory.s()
            )
            order_chain.apply_async()

// Batch Signal Processing
from django.db.models.signals import post_save
import threading

class SignalBatcher:
    queue = []
    lock = threading.Lock()
    
    @classmethod
    def batch_update(cls, model_instance):
        with cls.lock:
            cls.queue.append(model_instance)
            
            if len(cls.queue) >= 100:
                cls.flush()
    
    @classmethod
    def flush(cls):
        # Process 100 items at once instead of individually
        instances = cls.queue[:]
        cls.queue.clear()
        bulk_process(instances)

@receiver(post_save, sender=Book)
def batch_book_updates(sender, instance, **kwargs):
    SignalBatcher.batch_update(instance)