Async Task Processing with Celery

from celery import shared_task, chain, chord
from celery.exceptions import MaxRetriesExceededError
import structlog

logger = structlog.get_logger()

@shared_task(
    bind=True,
    queue="sap",
    max_retries=5,
    autoretry_for=(SAPConnectionError, SAPTimeoutError),
    retry_backoff=True,
    retry_backoff_max=600,
    retry_jitter=True,
)
def sync_order_to_sap(self, order_id: str):
    """Sync order to SAP ERP with automatic retry and idempotency."""
    log = logger.bind(order_id=order_id, task_id=self.request.id)
    
    try:
        # Idempotency check
        if sap_service.order_exists(order_id):
            log.info("Order already synced to SAP, skipping")
            return {"status": "already_synced"}
        
        order = order_repository.get(order_id)
        result = sap_service.create_order(order)
        
        # Update status
        order.sap_order_number = result.order_number
        order.sap_synced_at = datetime.utcnow()
        order_repository.save(order)
        
        log.info("Order synced to SAP", sap_order=result.order_number)
        return {"status": "synced", "sap_order": result.order_number}
        
    except MaxRetriesExceededError:
        log.error("Max retries exceeded for SAP sync")
        alert_ops("SAP sync failed permanently", order_id=order_id)
        raise

from celery import chain, chord, group
from tasks.order_tasks import (
    validate_order, reserve_inventory, sync_order_to_sap,
    send_confirmation_email, generate_invoice_pdf, update_analytics
)

def process_order_workflow(order_id: str):
    """
    Order processing workflow using Celery primitives.
    
    Flow:
    1. Validate order (must complete first)
    2. Reserve inventory (depends on validation)
    3. Sync to SAP (depends on inventory)  
    4. Parallel: email + PDF + analytics (after SAP sync)
    """
    workflow = chain(
        # Sequential dependencies
        validate_order.s(order_id),
        reserve_inventory.s(),
        sync_order_to_sap.s(),
        
        # Parallel processing after sync
        chord(
            group(
                send_confirmation_email.s(),
                generate_invoice_pdf.s(),
                update_analytics.s(),
            ),
            finalize_order.s(order_id)
        )
    )
    
    # Execute workflow asynchronously
    return workflow.apply_async()


# API endpoint stays fast
@router.post("/orders")
async def create_order(order: OrderCreate):
    # Quick validation and persist
    order_record = await order_service.create(order)
    
    # Queue background processing
    process_order_workflow.delay(order_record.id)
    
    # Return immediately (~200ms)
    return {"order_id": order_record.id, "status": "processing"}