Get training job results.
Source code in src/xpdeep/utils/jobs/model_training.py
| @initialized_client_verification
@initialized_project_verification
def get_job_result(job_id: UUID) -> TrainedModelArtifact:
"""Get training job results."""
try:
logging.print_job_progress(job_id)
except KeyboardInterrupt:
handle_api_validation_errors(
cancel_job.sync(Project.CURRENT.get().model.id, job_id, client=ClientFactory.CURRENT.get()())
)
raise
ongoing_job = handle_api_validation_errors(
get_one_job.sync(Project.CURRENT.get().model.id, job_id, client=ClientFactory.CURRENT.get()()),
)
if ongoing_job.status == JobStatus.ERROR:
remote_error = cast(JobModelResultsType0, ongoing_job.results)
error_message = f"{remote_error['error_class']} : {remote_error['error_detail']}"
raise ApiError(error_message)
if ongoing_job.results is None:
message = "Job failed"
raise ApiError(message)
if ongoing_job.type_ != JobType.TRAIN:
message = f"Unexpected Job type {ongoing_job.type_}."
raise ApiError(message)
return TrainedModelArtifact.from_dict(
handle_api_validation_errors(
get_one_trained_model.sync(
Project.CURRENT.get().model.id,
ongoing_job.results["id"],
client=ClientFactory.CURRENT.get()(),
),
).to_dict()
)
|