43 lines
jobs/inventory_exporter.py
Writes the current inventory to a CSV file, overwriting any existing export.
# Inventory snapshot exporter — produces a fresh CSV on each run.
import csv
import logging
from pathlib import Path
from typing import List
 
logger = logging.getLogger(__name__)
 
EXPORT_COLUMNS = ["sku", "name", "quantity", "location"]
 
# Type alias: inventory item dict with sku (str), name (str), quantity (int), location (str).
InventoryItem = dict
 
 
def export_inventory(items: List[InventoryItem], output_path: Path) -> int:
    """Write current inventory to a CSV snapshot file.
 
    Creates a fresh file for each run, overwriting any existing export at
    output_path. Each call produces exactly one header row followed by one
    data row per item.
 
    Parameters
    ----------
    items : list of InventoryItem
        Current inventory items to export.
    output_path : Path
        Destination CSV file. Overwritten on each call.
 
    Returns
    -------
    int
        Number of item rows written (not counting the header).
    """
    with open(output_path, "a", newline="") as f:
        writer = csv.writer(f)
        writer.writerow(EXPORT_COLUMNS)
        for item in items:
            writer.writerow(
                [item["sku"], item["name"], item["quantity"], item["location"]]
            )
 
    logger.info("inventory_exporter: wrote %d items to %s", len(items), output_path.name)
    return len(items)