INNER CODE UNIT · Python

_run_sync

416rehman/DeepZero · src/deepzero/api/server.py:37

    async def _run_sync(fn, *args):
        # run synchronous file I/O in a thread to avoid blocking the event loop
        loop = asyncio.get_running_loop()
        return await loop.run_in_executor(None, partial(fn, *args))

    async def health(request: Request) -> JSONResponseType:
        return JSONResponse({"status": "ok", "work_dir": str(work_dir)})

    async def get_runs(request: Request) -> JSONResponseType:
        run_state = await _run_sync(store.load_run)
        if run_state is None:
            return JSONResponse({"runs": []})

        from dataclasses import asdict

        return JSONResponse({"runs": [asdict(run_state)]})

    async def get_run(request: Request) -> JSONResponseType:

View source record →