python / expert
Snippet
Django Custom Management Command mit Async Fortschrittsanzeige
Dieses Snippet zeigt die Erstellung eines Django Management Commands, das Bestellungen asynchron mit Echtzeit-Fortschrittsverfolgung verarbeitet. Das Command verwendet Pythons asyncio-Modul, um mehrere Bestellungen concurrent zu verarbeiten und bietet visuelles Feedback durch eine Fortschrittsanzeige. Wichtige Features sind Batch-Verarbeitung für effizientes Speichermanagement, Dry-Run-Modus für sicheres Testen und ordnungsgemäße Datenbank-Transaktionsbehandlung.
snippet.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
import asynciofrom django.core.management.base import BaseCommandfrom django.db import transactionfrom myapp.models import Order, OrderItemclass Command(BaseCommand):help = 'Process pending orders asynchronously with progress tracking'def add_arguments(self, parser):parser.add_argument('--batch-size', type=int, default=100)parser.add_argument('--dry-run', action='store_true')async def process_order_batch(self, orders):async_tasks = []for order in orders:task = asyncio.create_task(self.process_single_order(order))async_tasks.append(task)return await asyncio.gather(*async_tasks)async def process_single_order(self, order):await asyncio.sleep(0.1)return {'order_id': order.id, 'status': 'processed'}def handle(self, *args, **options):batch_size = options['batch_size']is_dry_run = options['dry_run']pending_orders = Order.objects.filter(status='pending').select_related('customer')total = pending_orders.count()self.stdout.write(f'Processing {total} orders...')for i in range(0, total, batch_size):batch = list(pending_orders[i:i + batch_size])results = asyncio.run(self.process_order_batch(batch))progress = min(i + batch_size, total)percent = (progress / total) * 100self.stdout.write(f'Progress: {progress}/{total} ({percent:.1f}%)')self.stdout.write(self.style.SUCCESS(f'Completed processing {total} orders'))
django
Erklärung
1
from django.core.management.base import BaseCommand
Importiere BaseCommand um eigene Management Commands nach Djangos Command-Struktur zu erstellen
2
async def process_order_batch(self, orders)
Definiere async Funktion zur parallelen Verarbeitung mehrerer Bestellungen mit asyncio.gather()
3
asyncio.create_task(self.process_single_order(order))
Erstelle einzelne Tasks für jede Bestellung um gleichzeitige Verarbeitung im Batch zu ermöglichen
4
asyncio.run(self.process_order_batch(batch))
Führe den async Batch-Prozessor aus der synchronen handle() Methode mit asyncio.run() aus
5
--dry-run', action='store_true'
Füge Boolean-Flag Argument hinzu das Testen des Commands ohne Datenbankänderungen erlaubt