Snippet

Isolated Exception Handling in Asynchronous Signal Handlers

Extending Django Signal dispatching allows executing async and sync subscribers safely within event loops while isolating exceptions per receiver to prevent signal cascading failures.

snippet.py

python

import asyncio
from typing import Any, Callable, List
from asgiref.sync import sync_to_async
from django.dispatch import Signal
 
class AsyncIsolatedSignal(Signal):
    async def send_robust_async(self, sender: Any, **named: Any) -> List[tuple[Callable, Any]]:
        responses = []
        for receiver in self._live_receivers(sender):
            try:
                if asyncio.iscoroutinefunction(receiver):
                    res = await receiver(sender=sender, **named)
                else:
                    res = await sync_to_async(receiver, thread_sensitive=True)(sender=sender, **named)
                responses.append((receiver, res))
            except Exception as err:
                responses.append((receiver, err))
        return responses

django

Breakdown

class AsyncIsolatedSignal(Signal):

Custom Signal subclass providing asynchronous execution capabilities.

async def send_robust_async(self, sender: Any, **named: Any):

Asynchronous dispatch method iterating over registered signal receivers.

res = await sync_to_async(receiver, thread_sensitive=True)(...)

Wraps traditional synchronous receivers into thread-sensitive async coroutines.

except Exception as err:

Catches individual receiver errors without aborting execution for remaining receivers.

Previous snippet Next snippet

More Expert Python exercises →

Isolated Exception Handling in Asynchronous Signal Handlers

Related