Print all the progress messages for a job in real-time.
Source code in src/xpdeep/utils/jobs.py
| @initialized_client_verification
@initialized_project_verification
def print_job_progress(job_id: str) -> None:
"""Print all the progress messages for a job in real-time."""
with (
ClientFactory.CURRENT.get()() as client,
connect_sse(
client.get_httpx_client(),
"GET",
f"/{Project.CURRENT.get().model.id}/jobs/{job_id}/progress",
headers={"Accept-Encoding": ""},
) as event_source,
):
try:
for event in event_source.iter_sse():
print(json.loads(event.data)["console_output"], end="") # noqa: T201
except SSEError:
event_source.response.read()
warnings.warn(
f'Cannot display progress bar. {event_source.response.json()["detail"]}', Warning, stacklevel=1
)
|