python / expert
Snippet
Asynchrone QuerySet-Streaming-Verarbeitung mit Kontextueller Fehlerbehandlung
Die asynchrone Iteration über Django-QuerySets mittels `aiter()` ermöglicht nicht-blockierendes Datenbank-Streaming, kombiniert mit verschachtelten Try-Except-Blöcken zur Behandlung einzelner Elementfehler vs. Datenbankfehler.
snippet.py
python
1
2
3
4
5
6
7
8
9
10
11
12
from django.db import modelsfrom typing import AsyncGeneratorasync def stream_audit_logs(queryset: models.QuerySet) -> AsyncGenerator[models.Model, None]:try:async for record in queryset.aiter():try:yield recordexcept Exception as item_err:continueexcept Exception as stream_err:raise RuntimeError(f'Stream interrupted: {stream_err}') from stream_err
django
Erklärung
1
async for record in queryset.aiter():
Iteriert asynchron über ein Django-QuerySet unter Verwendung des asynchronen Generator-Protokolls.
2
yield record
Gibt einzelne Modellinstanzen an den asynchronen Aufrufer-Kontext zurück.
3
except Exception as item_err:
Fängt Fehler bei der Verarbeitung einzelner Elemente ab, um das Streaming fortzusetzen.