(checkpoint_name: str, req: UpdateRequest)
| 94 | |
| 95 | @app.post("/v1/checkpoints/{checkpoint_name}/update") |
| 96 | async def update(checkpoint_name: str, req: UpdateRequest) -> Response: |
| 97 | def update_func(socket_paths: list[tuple[str, str]]): |
| 98 | if req.update_url is None: |
| 99 | return |
| 100 | if req.inference_group_ranks: |
| 101 | socket_paths = [socket_paths[i] for i in req.inference_group_ranks] |
| 102 | request_inference_to_update( |
| 103 | req.update_url, dict(socket_paths), timeout=req.timeout, uds=req.uds |
| 104 | ) |
| 105 | |
| 106 | return wrap_exception(lambda: ps.update(checkpoint_name, update_func, ranks=req.ranks)) |
| 107 | |
| 108 | return app |
nothing calls this directly
no test coverage detected