python / expert
Snippet
Asynchronous QuerySet Streaming with Contextual Exception Recovery
Asynchronous iteration over Django QuerySets using `aiter()` enables non-blocking database streaming, combined with nested try-except blocks to gracefully handle individual element failures versus database stream failures.
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
Breakdown
1
async for record in queryset.aiter():
Iterates asynchronously over a Django QuerySet using asynchronous generator protocol.
2
yield record
Yields individual model instances to the async caller context.
3
except Exception as item_err:
Catches mid-stream consumer processing errors to allow stream continuation.