python / expert
Snippet
Isolierte Fehlerbehandlung in asynchronen Signal-Handlern
Die Erweiterung der Django-Signalverarbeitung ermöglicht die sichere Ausführung asynchroner und synchroner Empfänger in Event-Loops und isoliert Ausnahmen pro Empfänger, um Kaskadenfehler zu verhindern.
snippet.py
python
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
import asynciofrom typing import Any, Callable, Listfrom asgiref.sync import sync_to_asyncfrom django.dispatch import Signalclass 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
Erklärung
1
class AsyncIsolatedSignal(Signal):
Benutzerdefinierte Signal-Unterklasse mit asynchronen Ausführungsfunktionen.
2
async def send_robust_async(self, sender: Any, **named: Any):
Asynchrone Versandmethode, die registrierte Signalempfänger nacheinander aufruft.
3
res = await sync_to_async(receiver, thread_sensitive=True)(...)
Wandelt traditionelle synchrone Empfänger in thread-sensitive asynchrone Coroutinen um.
4
except Exception as err:
Fängt Fehler einzelner Empfänger ab, ohne die Ausführung für übrige Empfänger abzubrechen.