Building Async Batch Processors for Inventory Updates

import asyncio
import hashlib
import logging
import json
from datetime import datetime, timezone
from pathlib import Path
from typing import List, Dict, Any
from dataclasses import dataclass, field
import pandas as pd
from pydantic import BaseModel, ValidationError, field_validator, ConfigDict
import httpx

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s | %(levelname)s | %(name)s | %(message)s",
    handlers=[
        logging.FileHandler("inventory_audit.log"),
        logging.StreamHandler()
    ]
)
logger = logging.getLogger("compliance.inventory_processor")

class InventoryItem(BaseModel):
    model_config = ConfigDict(extra="forbid")
    asset_id: str
    grant_code: str
    category: str
    quantity: int
    location: str
    acquisition_date: str

    @field_validator("grant_code")
    @classmethod
    def validate_grant_prefix(cls, v: str) -> str:
        allowed = ("NIH-", "NSF-", "DOE-", "DOD-", "EPA-")
        if not v.startswith(allowed):
            raise ValueError(f"Grant code must begin with institutional prefix: {allowed}")
        return v.upper()

    @field_validator("quantity")
    @classmethod
    def validate_non_negative(cls, v: int) -> int:
        if v < 0:
            raise ValueError("Quantity must be non-negative per OSHA/EPA manifest standards")
        return v

@dataclass
class BatchResult:
    committed: List[Dict[str, Any]] = field(default_factory=list)
    quarantined: List[Dict[str, Any]] = field(default_factory=list)
    batch_hash: str = ""
    processed_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))

class AsyncInventoryProcessor:
    def __init__(self, api_base_url: str, chunk_size: int = 500, max_retries: int = 3):
        self.api_base_url = api_base_url.rstrip("/")
        self.chunk_size = chunk_size
        self.max_retries = max_retries
        self._idempotency_store: set[str] = set()

    def _compute_batch_hash(self, records: List[Dict[str, Any]]) -> str:
        payload = json.dumps(records, sort_keys=True).encode()
        return hashlib.sha256(payload).hexdigest()

    def _generate_idempotency_key(self, record: Dict[str, Any]) -> str:
        return f"{record.get('asset_id')}:{record.get('grant_code')}"

    async def _commit_chunk(self, chunk: List[Dict[str, Any]], client: httpx.AsyncClient) -> List[Dict[str, Any]]:
        idem_key = self._compute_batch_hash(chunk)
        headers = {"Idempotency-Key": idem_key, "X-Compliance-Standard": "NIH-NSF-OSHA-EPA"}
        
        for attempt in range(1, self.max_retries + 1):
            try:
                async with client.stream("POST", f"{self.api_base_url}/inventory/upsert", json=chunk, headers=headers) as resp:
                    resp.raise_for_status()
                    return chunk
            except httpx.HTTPStatusError as e:
                if 500 <= e.response.status_code < 600:
                    logger.warning(f"Transient server error (attempt {attempt}/{self.max_retries}): {e.response.status_code}")
                    await asyncio.sleep(2 ** attempt)
                    continue
                raise
            except httpx.RequestError:
                logger.warning(f"Network drop (attempt {attempt}/{self.max_retries}). Retrying...")
                await asyncio.sleep(2 ** attempt)
                continue
        raise RuntimeError("Max retries exceeded for chunk commit")

    async def _process_chunk(self, chunk: List[Dict[str, Any]], client: httpx.AsyncClient) -> BatchResult:
        result = BatchResult()
        valid_records = []
        
        for record in chunk:
            key = self._generate_idempotency_key(record)
            if key in self._idempotency_store:
                logger.debug(f"Skipping duplicate idempotency key: {key}")
                continue

            try:
                validated = InventoryItem(**record)
                valid_records.append(validated.model_dump())
                self._idempotency_store.add(key)
            except ValidationError as ve:
                result.quarantined.append({
                    "original": record,
                    "errors": ve.errors(),
                    "reason": "schema_violation"
                })

        if valid_records:
            try:
                committed = await self._commit_chunk(valid_records, client)
                result.committed.extend(committed)
            except Exception as e:
                logger.error(f"Chunk commit failed. Routing to deterministic fallback: {e}")
                for rec in valid_records:
                    result.quarantined.append({
                        "original": rec,
                        "errors": [{"msg": "Network/commit failure during chunk submission"}],
                        "reason": "transient_failure"
                    })

        result.batch_hash = self._compute_batch_hash(valid_records)
        return result

    async def process_manifest(self, file_path: Path) -> List[BatchResult]:
        logger.info(f"Starting async batch processing for: {file_path}")
        df = pd.read_csv(file_path, dtype=str)
        records = df.to_dict(orient="records")
        chunks = [records[i:i + self.chunk_size] for i in range(0, len(records), self.chunk_size)]

        results = []
        async with httpx.AsyncClient(timeout=30.0) as client:
            tasks = [self._process_chunk(chunk, client) for chunk in chunks]
            results = await asyncio.gather(*tasks, return_exceptions=True)

        final_results = []
        for r in results:
            if isinstance(r, Exception):
                logger.error(f"Chunk processing failed with exception: {r}")
                final_results.append(BatchResult(quarantined=[{"error": str(r)}]))
            else:
                final_results.append(r)

        total_committed = sum(len(r.committed) for r in final_results)
        total_quarantined = sum(len(r.quarantined) for r in final_results)
        logger.info(f"Completed processing. Committed: {total_committed} | Quarantined: {total_quarantined}")
        return final_results