"""Endpoint for checking job status in a study."""
from typing import Any, Optional
from imednet.core.endpoint.mixins import EdcListPathGetEndpoint
from imednet.core.paginator import AsyncJsonListPaginator, JsonListPaginator
from imednet.models.jobs import JobStatus
[docs]class JobsEndpoint(EdcListPathGetEndpoint[JobStatus]):
"""
API endpoint for retrieving status and details of jobs in an iMedNet study.
This endpoint provides methods to fetch individual job status by batch ID
and list all jobs for a study.
"""
PATH = "jobs"
MODEL = JobStatus
PAGINATOR_CLS = JsonListPaginator
ASYNC_PAGINATOR_CLS = AsyncJsonListPaginator
def _raise_not_found(self, study_key: Optional[str], item_id: Any) -> None:
raise ValueError(f"Job {item_id} not found in study {study_key}")
[docs] def get(self, study_key: Optional[str], batch_id: str) -> JobStatus:
"""
Get a specific job by batch ID.
This method performs a direct API request using the provided
``batch_id``; it does not use caching or the ``refresh`` flag.
Args:
study_key: Study identifier
batch_id: Batch ID of the job
Returns:
JobStatus object with current state and timestamps
Raises:
ValueError: If the job is not found
"""
return super().get(study_key, batch_id)
[docs] async def async_get(self, study_key: Optional[str], batch_id: str) -> JobStatus:
"""
Asynchronously get a specific job by batch ID.
This is the async variant of :meth:`get`. Like the sync version,
it issues a direct request by ``batch_id`` without any caching.
Args:
study_key: Study identifier
batch_id: Batch ID of the job
Returns:
JobStatus object with current state and timestamps
Raises:
ValueError: If the job is not found
"""
return await super().async_get(study_key, batch_id)