|

@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.