Bulk Operations — Aquilia Documentation
Comprehensive guide and documentation for Bulk Operations in the Aquilia framework. View API reference, examples, and implementation patterns.
Docs / Models / Bulk Operations Bulk Operations Efficient batch INSERT, bulk UPDATE, bulk DELETE, chunked iteration, and upsert patterns. Signal behavior: bulk_create(), .update(), and .delete() do not fire pre_save, post_save, pre_delete, or post_delete signals. They operate directly at the SQL layer. Use instance-level .save() / .delete_instance() when signals are required. bulk_create() Inserts a list of model instances in one or more batched INSERT statements. Dramatically faster than calling .save() per instance for large datasets. # Signature async def bulk_create( db: AquiliaDatabase, objs: list[Model], batch_size: int = 1000, ) -> None # Example: import 50,000 products products = [ Product(name=row['name'], price=row['price'], sku=row['sku']) for row in csv_rows ] await Product.objects.bulk_create(db, products, batch_size=500) # → Executes 100 INSERT statements, 500 rows each # After bulk_create, PKs may or may not be assigned depending on backend. # Re-query if you need the assigned IDs: created = await Product.objects.filter(sku__in=[p.sku for p in products]).all() batch_size guidance: SQLite has a default variable limit of 999 per statement. Keep batch_size at or below 999 for SQLite. PostgreSQL handles much larger batches efficiently. MySQL performs well at 500–5000 rows per batch depending on row width. .update() — Bulk UPDATE Issues a single UPDATE ... SET ... WHERE ... statement for all rows matching the queryset filter. Accepts literal values, F() expressions, arithmetic, and Case() expressions. # Signature async def update(self, db: AquiliaDatabase, **kwargs) -> int # returns rows affected # Literal update — set all active users' role to 'member' count = await User.objects.filter(is_active=True).update(db, role='member') # F() expression — atomic increment (no Python read-modify-write) await Product.objects.filter(id__in=popular_ids).update(db, views=F('views') + 1) # Arithmetic on multiple fields await Order.objects.filter(status='pending').update( db, total=F('subtotal') + F('tax'), updated_at=Now(), ) # Conditional update with Case/When from aquilia.models.expression import Case, When, Value await Product.objects.update( db, status=Case( When(stock=0, then=Value('out_of_stock')), When(stock__lt=10, then=Value('low_stock')), default=Value('in_stock'), ) ) # Limit scope with filter affected = await User.objects.filter( last_login__lt=cutoff_date ).update(db, is_active=False) .delete() — Bulk DELETE Issues a single DELETE FROM ... WHERE ... for all rows matching the queryset. No signals fire. # Signature async def delete(self, db: AquiliaDatabase) -> int # returns rows deleted # Delete all expired sessions deleted = await Session.objects.filter(expires_at__lt=now).delete(db) # Delete with complex filter await Log.objects.filter( created_at__lt=cutoff, level__in=['DEBUG', 'INFO'], ).delete(db) # WARNING: .delete() with no filter deletes ALL rows in the table. # Always verify your filter scope before calling .delete(). in_bulk() Fetches a list of PKs and returns a dict mapping each PK to its model instance. Internally chunks large lists to avoid hitting SQLite's variable limit. # Signature async def in_bulk( self, id_list: list, batch_size: int = 999, ) -> dict[Any, Model] # Fetch multiple users by ID without N+1 user_ids = [1, 5, 42, 99] user_map = await User.objects.in_bulk(db, user_ids) # → alice = user_map[1] # Large ID list — automatically chunked into batches of 999 all_ids = list(range(1, 50001)) product_map = await Product.objects.in_bulk(db, all_ids) # → Makes ceil(50000/999) queries, merges results into one dict Upsert Patterns Aquilia provides three methods for common get-or-create / upsert patterns. Understand their guarantees before choosing: Method Creates? Updates? Race-safe? ))} RuntimeWarning: get_or_create() and update_or_create() emit a RuntimeWarning on every call, since both are a plain SELECT-then-INSERT/UPDATE and not atomic under concurrent access. Prefer find_or_create() when a unique constraint is available — it does not warn. get_or_create() # Returns (instance, created: bool) user, created = await User.objects.get_or_create( db, defaults= , email='alice@example.com', # lookup fields ) if created: print("New user created") else: print("Existing user fetched") update_or_create() # Returns (instance, created: bool) # Updates existing row with defaults fields, or creates new row profile, created = await Profile.objects.update_or_create( db, defaults= , user_id=42, # lookup key ) find_or_create() — Atomic Upsert Uses INSERT ... ON CONFLICT DO NOTHING under the hood, eliminating the TOCTOU race condition present in get_or_create. Safe under concurrent load without transactions. # Returns (instance, created: bool) tag, created = await Tag.objects.find_or_create( db, defaults= , create_defaults= , name='Python', # lookup + unique constraint field ) # Under the hood: # INSERT INTO "tags" ("name", "color", "slug") # VALUES (?, ?, ?) ON CONFLICT DO NOTHING # Then SELECT to return the row regardless of insert vs conflict .iterator() — Memory-Efficient Chunked Iteration .iterator() returns an async generator that fetches and yields rows in chunks. Unlike .all(), it does not load the entire result set into memory at once — essential for processing hundreds of thousands of rows. # Signature def iterator(self, chunk_size: int = 2000) -> AsyncGenerator[Model, None] # Process 1M rows without OOM async for user in User.objects.filter(is_active=True).iterator(chunk_size=500): await process_user(user) # With ordering — ensures consistent chunking async for order in Order.objects.order('id').iterator(): await send_receipt(order) # Cannot use .all() or .first() after .iterator() — it is a terminal method # Cannot slice or further chain after calling .iterator() select_for_update() Adds SELECT ... FOR UPDATE locking to the query. Must be used inside an atomic() block. Not supported on SQLite (raises QueryFault on SQLite). # Signature def select_for_update( self, nowait: bool = False, skip_locked: bool = False, ) -> QuerySet from aquilia.models import atomic async with atomic(db): # Lock the row for the duration of the transaction account = await ( BankAccount.objects .filter(id=account_id) .select_for_update() .first() ) account.balance -= amount await account.save(db) # nowait=True → raises immediately if row is locked (no waiting) async with atomic(db): try: job = await Job.objects.filter(status='pending').select_for_update(nowait=True).first() except Exception: return # Another worker has it # skip_locked=True → skips locked rows (queue worker pattern) async with atomic(db): job = await ( Job.objects .filter(status='pending') .select_for_update(skip_locked=True) .order('created_at') .first() ) if job: await process_job(job) Recursive CTEs Advanced Usage );
Go to Homepage