Products - Celery Tasks

Location: api/nextango/apps/products/tasks.py Last Updated: 2026-07-06

Overview

Background tasks for the products domain. Three task families use the same bounded fan-out pattern: a dispatcher resolves the work set, splits it into chunks, and fires a Celery chord whose header tasks process chunks in parallel and whose finalizer folds the per-chunk summaries into one result. This keeps per-store and per-variant work parallel across workers instead of serial in one task, so the pattern holds from one store to thousands.

The families:

• Filter manifest generation - periodic, every 15 minutes, pre-builds per-store filter manifests
• Daily inventory snapshot rollup - periodic, nightly, writes DailyInventorySnapshot aggregates per store
• Inventory reconciliation - event-driven, reconciles InventoryLevel records after variant syncs and new-store creation

Task Summary

Filter Manifest Generation

generate_filter_manifests

Trigger: Celery Beat, every 15 minutes.

Dispatcher. Resolves the active store IDs, splits them into chunks of 50, and fires a chord of generate_filter_manifests_chunk tasks with generate_filter_manifests_finalize as the body. Per-store failures stay isolated in the chunk workers; the dispatcher's retry covers dispatch-level failures such as resolving the active-store set, retrying up to three times with a five-minute countdown.

Returns a dispatch summary:

{'success': True, 'stores': <int>, 'chunks': <int>, 'message': 'Manifests generation dispatched'}

With no active stores it returns early with stores: 0.

generate_filter_manifests_chunk

Chunk worker. For each store in the chunk, calls generate_store_manifest (see Smart Filter Service) and writes the manifest to MEDIA_ROOT/manifests/store_{store_id}.json. Manifests capture the current set of available filter values for each store's catalog; the frontend reads these pre-built JSON files to render filter panels without per-request database queries. A single store failure never raises out of the chunk, so the chord finalizer always runs.

Returns {'generated': <int>, 'failed': <int>, 'chunk_size': <int>}.

generate_filter_manifests_finalize

Chord finalizer. Folds the per-chunk summaries into one result:

{'success': True, 'generated': <int>, 'failed': <int>, 'chunks': <int>}

Daily Inventory Snapshot Rollup

rollup_daily_inventory_snapshot

Task name: products.rollup_daily_inventory_snapshot Trigger: Celery Beat, daily at 01:00 UTC. Also callable with an explicit target_date (defaults to today).

Dispatcher. Resolves active store IDs, splits them into chunks of 50, and fires a chord of rollup_daily_inventory_snapshot_chunk tasks with rollup_daily_inventory_snapshot_finalize as the body. The rollup is idempotent: the chunk worker writes via update_or_create, so re-running for the same date overwrites rather than duplicates.

Under eager execution (tests) the call returns the finalizer's dict; in production it returns an AsyncResult. The settled result shape:

{'date': '<ISO date>', 'stores_processed': <int>, 'errors': <int>}

rollup_daily_inventory_snapshot_chunk

Task name: products.rollup_daily_inventory_snapshot_chunk

Chunk worker. For each store in the chunk, aggregates the store's active InventoryLevel rows and writes one DailyInventorySnapshot per (store, date):

Per-store failures increment a local error counter and the loop continues, so the chord finalizer runs deterministically.

rollup_daily_inventory_snapshot_finalize

Task name: products.rollup_daily_inventory_snapshot_finalize

Chord finalizer. Folds chunk summaries into the dispatcher-level return contract above, preserving the shape existing callers and tests read.

Inventory Reconciliation

reconcile_product_inventory_dispatch

Task name: products.reconcile_product_inventory_dispatch Trigger: Dispatched on transaction commit by Product.sync_variants_to_model_records(dispatch_inventory_reconcile_async=True), the webhook-path variant sync. Non-webhook callers (admin actions, views, services, serializer writes) reconcile inline instead.

Dispatcher. Given a product and a snapshot of its active variant IDs, it builds the union of each variant's target stores (all active stores, or the variant's store_location refs for store-specific variants) plus the stores that already hold an InventoryLevel for those variants (potential soft-delete targets). It splits the store IDs into chunks of 50 and fires a chord of reconcile_inventory_levels_chunk tasks with reconcile_inventory_levels_finalize as the body.

The reconcile is idempotent: the per-(variant, store) helpers use get_or_create and is_active gates, so re-running converges to the same state. A missing product or empty variant set returns a zeroed result.

The settled result shape (finalizer dict under eager execution, AsyncResult in production):

{
    'product_id': str,
    'created': int,          # InventoryLevel rows created
    'reactivated': int,      # inactive rows reactivated for target stores
    'sku_updated': int,      # variant_sku propagated after a SKU change
    'noop_active': int,      # already correct, untouched
    'soft_deleted': int,     # deactivated for stores no longer targeted
    'noop_inactive': int,    # already inactive, untouched
    'digital_skipped': int,  # digital variants, no tracking created
    'digital_soft_deleted': int,
    'errors': int,
}

reconcile_inventory_levels_chunk

Task name: products.reconcile_inventory_levels_chunk

Chunk worker. Iterates the (variant, store) pairs in its chunk. For a target store it reconciles the level (create, reactivate, or propagate a SKU change); for a non-target store that still holds a level it soft-deletes. Existing levels resolve both by variant FK and by the (product, variant_sku) natural key, which catches rows created by the inventory-seeding signal before a variant FK was linked, and rows whose SKU changed. Per-pair failures increment a local error counter and never raise out of the chunk.

reconcile_inventory_levels_finalize

Task name: products.reconcile_inventory_levels_finalize

Chord finalizer. Folds chunk counts into the canonical return shape above.

New-Store Inventory Backfill

reconcile_new_store_inventory

Task name: products.reconcile_new_store_inventory Trigger: Dispatched on transaction commit by the stores app's Store-create signal, on creation only.

A store created after products publish would otherwise hold zero InventoryLevel rows and show as empty. This task closes that gap. It selects the variants that belong at the new store: available-everywhere variants (store_specific_product=False) plus store-specific variants whose store_location references the store, excluding variants soft-deleted from Sanity. It splits the variant IDs into chunks of 200 and dispatches one reconcile_new_store_inventory_chunk per chunk, targeting only the one new store rather than the all-active-stores fan-out the product-centric reconcile uses.

Returns {'store_id': str, 'dispatched_chunks': int, 'variants': int}. A missing store returns a zeroed result.

reconcile_new_store_inventory_chunk

Task name: products.reconcile_new_store_inventory_chunk

Chunk worker. For each variant in the chunk, creates the InventoryLevel at the new store through the per-store reconcile helper. Idempotent via the unique (product, variant_sku, store) constraint. Per-variant failures increment a local error counter and never raise out.

Returns {'store_id': str, 'processed': int, 'errors': int}.

Celery Beat Configuration

from celery.schedules import crontab

CELERY_BEAT_SCHEDULE = {
    'generate-filter-manifests': {
        'task': 'nextango.apps.products.tasks.generate_filter_manifests',
        'schedule': crontab(minute='*/15'),
    },
    'rollup-daily-inventory-snapshot': {
        'task': 'products.rollup_daily_inventory_snapshot',
        'schedule': crontab(hour=1, minute=0),
        'options': {'expires': 3600},
    },
}

Manual Execution

from nextango.apps.products.tasks import (
    generate_filter_manifests,
    rollup_daily_inventory_snapshot,
)

# Queue the manifest fan-out
generate_filter_manifests.delay()

# Re-run a snapshot rollup for a specific date (idempotent).
# target_date must be a datetime.date, so call the task directly
# from a shell rather than through .delay with a date argument.
import datetime
rollup_daily_inventory_snapshot(target_date=datetime.date(2026, 6, 11))

Manifests can also be generated via the management command:

docker exec nextango_api python manage.py generate_filter_manifests

Related Documentation

• Signals - inventory initialization and the signals that complement these tasks
• Models - Product, ProductVariant, InventoryLevel, DailyInventorySnapshot
• Services - inventory_valuation_expression and sync services
• Smart Filter Service - generate_store_manifest and manifest consumption
• Views - product catalog and inventory endpoints