python / expert
Snippet
Django Custom Management Command with Async Progress Tracking
This snippet demonstrates creating a Django management command that processes orders asynchronously with real-time progress tracking. The command uses Python's asyncio module to handle multiple orders concurrently while providing visual feedback through a progress bar. Key features include batch processing to manage memory efficiently, dry-run mode for safe testing, and proper database transaction handling.
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
Breakdown
1
from django.core.management.base import BaseCommand
Import BaseCommand to create custom management commands following Django's command structure
2
async def process_order_batch(self, orders)
Define async function to process multiple orders concurrently using asyncio.gather() for parallel execution
3
asyncio.create_task(self.process_single_order(order))
Create individual tasks for each order to enable concurrent processing within the batch
4
asyncio.run(self.process_order_batch(batch))
Execute the async batch processor from synchronous handle() method using asyncio.run()
5
--dry-run', action='store_true'
Add boolean flag argument that allows testing the command without modifying database state