Polling endpoint for real-time progress updates during generation.
(request: Request)
| 358 | |
| 359 | @app.get("/progress") |
| 360 | async def progress_poll(request: Request): |
| 361 | """Polling endpoint for real-time progress updates during generation.""" |
| 362 | session_id = request.query_params.get("session_id", "") |
| 363 | path = _progress_file(session_id) |
| 364 | try: |
| 365 | with open(path, 'r') as f: |
| 366 | data = json.load(f) |
| 367 | return JSONResponse(data) |
| 368 | except (FileNotFoundError, json.JSONDecodeError): |
| 369 | return JSONResponse({"stage": "Waiting...", "step": 0, "total": 0, "done": False}) |
| 370 | |
| 371 | @app.api() |
| 372 | @spaces.GPU(duration=30) |
nothing calls this directly
no test coverage detected