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 csvimport loggingfrom pathlib import Pathfrom typing import Listlogger = logging.getLogger(__name__)
EXPORT_COLUMNS = ["sku", "name", "quantity", "location"]# Type alias: inventory item dict with sku (str), name (str), quantity (int), location (str).InventoryItem = dictdef 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)