|

@router.get("/sse/{task_id}")
async def sse_stream(task_id: str):
  async def event_generator():
      queue = task_queues.get(task_id)
      while True:
          msg = await queue.get()
          if msg is None: break       # task done signal
          yield f"data: {msg}\n\n"
  return StreamingResponse(event_generator(), media_type="text/event-stream")

with ThreadPoolExecutor(max_workers=48) as executor:
  futures = {executor.submit(upload_single, p, bucket): p for p in paths}
  for future in as_completed(futures):
      try:
          future.result()
          task_queue.put(f"✓ {futures[future].name}")  # SSE push
      except Exception as e:
          failed.append((futures[future], str(e)))
rebuild_index(bucket)   # rebuild resource index

Core principle: the view layer only renders; business logic lives in Composables. Four Composables each own their domain and compose with each other — useOrderTask internally composes useSSETask, reusing log-stream logic without duplication. TypeScript unified request/response types across 10 API modules, catching field-rename errors at compile time during multiple refactors.

SQLModel (SQLAlchemy + Pydantic fusion) lets the same Model class serve as both ORM mapping and API schema, eliminating redundant definitions. Alembic manages PostgreSQL migration history so every schema change is versioned. The three datasources were chosen for their access patterns — not over-engineering, but the right tool for each job.

The PLC layer uses python-snap7 to connect to a Siemens S7-200 SMART controller. It supports connection testing, output-point testing, and order-line-based pulse signals. The device-communication layer is separated from the order model so future integrations can swap in serial, Modbus, or HTTP device interfaces without changing the master system.

Early scripts had parameters scattered through code: changing a store, template, or output directory meant editing scripts. The current system extracts task definition from execution logic. settings.yaml describes default paths and globals, catalog.yaml merges product catalog metadata, and upload_plan.json orchestrates date-based batch tasks. The same backend pipeline exposes both CLI and HTTP APIs: command line for development/debugging, frontend buttons for operations.