Rapidata job

RapidataJob

RapidataJob(
    job_id: str,
    name: str,
    audience_id: str,
    created_at: datetime,
    definition_id: str,
    openapi_service: OpenAPIService,
    pipeline_id: str | None = None,
)

An instance of a Rapidata job.

Used to interact with a specific job in the Rapidata system, such as getting status and retrieving results.

A job is created from a job definition and an audience, and represents a specific run of that definition.

Parameters:

Name	Type	Description	Default
job_id	str	The ID of the job.	required
name	str	The name of the job.	required
openapi_service	OpenAPIService	The OpenAPIService instance for API interaction.	required

Source code in src/rapidata/rapidata_client/job/rapidata_job.py

def __init__(
    self,
    job_id: str,
    name: str,
    audience_id: str,
    created_at: datetime,
    definition_id: str,
    openapi_service: OpenAPIService,
    pipeline_id: str | None = None,
):
    self.id = job_id
    self.name = name
    self.audience_id = audience_id
    self._openapi_service = openapi_service
    self.created_at = created_at
    self.definition_id = definition_id
    self.__pipeline_id = pipeline_id
    self.__completed_at = None
    self.__estimated_cost: CostEstimate | None = None
    self.job_details_page = f"https://app.{self._openapi_service.environment}/audiences/{self.audience_id}/job/{self.id}"
    logger.debug("RapidataJob initialized")

completed_at property

completed_at: datetime | None

Returns the completion date of the job, or None if not completed.

pipeline_id property

pipeline_id: str

Returns the pipeline ID of the job.

estimated_cost property

estimated_cost: CostEstimate

An approximate estimate of what this job will cost to run to completion.

This is an estimate, not the final bill - see :class:CostEstimate. The estimate is priced shortly after the job is created; this call waits for it to become available.

Raises:

Type	Description
TimeoutError	If the estimate is still not available after a few minutes.

get_status

get_status() -> str

Gets the status of the job.

Returns:

Type	Description
str	The current status of the job as a string.

Source code in src/rapidata/rapidata_client/job/rapidata_job.py

def get_status(self) -> str:
    """
    Gets the status of the job.

    Returns:
        The current status of the job as a string.
    """
    with tracer.start_as_current_span("RapidataJob.get_status"):
        return self._fetch_job().state.value

get_progress

get_progress() -> JobProgress

Gets a snapshot of how far along the job is.

Unlike meth:get_status, which only reports the coarse job state, this returns the labeling completion percentage and — for custom audiences — the recruiting funnel of the annotator pool behind the job, so you can tell normal labeling from a job that is waiting on recruiting or has stalled.

This does not block: it reports the current progress and returns immediately.

Returns:

Name	Type	Description
JobProgress	JobProgress	The current progress of the job.

Source code in src/rapidata/rapidata_client/job/rapidata_job.py

def get_progress(self) -> JobProgress:
    """Gets a snapshot of how far along the job is.

    Unlike :py:meth:`get_status`, which only reports the coarse job state, this
    returns the labeling completion percentage and — for custom audiences — the
    recruiting funnel of the annotator pool behind the job, so you can tell normal
    labeling from a job that is waiting on recruiting or has stalled.

    This does not block: it reports the current progress and returns immediately.

    Returns:
        JobProgress: The current progress of the job.
    """
    with tracer.start_as_current_span("RapidataJob.get_progress"):
        from rapidata.rapidata_client.job.progress import JobProgress

        job = self._fetch_job()
        workflow_progress = self._get_workflow_progress()

        return JobProgress(
            state=job.state.value,
            completion_percentage=(
                float(workflow_progress.completion_percentage)
                if workflow_progress
                else 0.0
            ),
            recruiting=self._get_recruiting_metrics(),
        )

get_results

get_results() -> RapidataResults

Gets the results of the job.

If the job is still processing, this method will block until the job is completed and then return the results. If the job's results have gone stale, regeneration is triggered automatically and this method blocks until the fresh results are ready.

Returns:

Name	Type	Description
RapidataResults	RapidataResults	The results of the job.

Raises:

Type	Description
Exception	If failed to get job results, or if the job cannot complete without intervention — it is in manual review (ManualApproval), spend-limited (SpendLimited), or assigned to an audience that can never graduate annotators (recruiting never started, or its pool is empty).

Source code in src/rapidata/rapidata_client/job/rapidata_job.py

def get_results(self) -> RapidataResults:
    """
    Gets the results of the job.

    If the job is still processing, this method will block until the job is
    completed and then return the results.
    If the job's results have gone stale, regeneration is triggered automatically
    and this method blocks until the fresh results are ready.

    Returns:
        RapidataResults: The results of the job.

    Raises:
        Exception: If failed to get job results, or if the job cannot complete
            without intervention — it is in manual review (``ManualApproval``),
            spend-limited (``SpendLimited``), or assigned to an audience that can
            never graduate annotators (recruiting never started, or its pool is
            empty).
    """
    with tracer.start_as_current_span("RapidataJob.get_results"):
        from rapidata.api_client.exceptions import ApiException
        from rapidata.rapidata_client.results.rapidata_results import (
            RapidataResults,
        )

        logger.info("Getting results for job '%s'...", self)

        # Stale results have no downloadable file until the pipeline is re-run;
        # trigger that automatically before waiting for the re-completion.
        if self.get_status() == "StaleResults":
            self._regenerate_results()

        self._wait_for_status(
            target_statuses=["Completed", "Failed"],
            status_message="Job '%s' is in status %s, waiting for completion...",
        )

        try:
            results = (
                self._openapi_service.order.job_api.job_job_id_download_results_get(
                    job_id=self.id
                )
            )
            return RapidataResults(json.loads(results))
        except (ApiException, json.JSONDecodeError) as e:
            raise Exception(f"Failed to get job results: {str(e)}") from e

display_progress_bar

display_progress_bar(refresh_rate: int = 5) -> None

Displays a progress bar for the job processing using tqdm.

Parameters:

Name	Type	Description	Default
refresh_rate	int	How often to refresh the progress bar, in seconds.	5

Raises:

Type	Description
ValueError	If refresh_rate is less than 1.
Exception	If the job has failed, or can't progress on its own — it is in ManualApproval or SpendLimited, or assigned to an audience that can never graduate annotators (recruiting never started, or its pool is empty).

Source code in src/rapidata/rapidata_client/job/rapidata_job.py

def display_progress_bar(self, refresh_rate: int = 5) -> None:
    """
    Displays a progress bar for the job processing using tqdm.

    Args:
        refresh_rate: How often to refresh the progress bar, in seconds.

    Raises:
        ValueError: If refresh_rate is less than 1.
        Exception: If the job has failed, or can't progress on its own — it is
            in ``ManualApproval`` or ``SpendLimited``, or assigned to an audience
            that can never graduate annotators (recruiting never started, or its
            pool is empty).
    """
    if refresh_rate < 1:
        raise ValueError("refresh_rate must be at least 1")

    job = self._fetch_job()
    if job.state == AudienceJobState.COMPLETED:
        managed_print(f"Job '{self}' is already completed.")
        return

    if job.state == AudienceJobState.FAILED:
        raise Exception(f"Job '{self}' has failed: {job.failure_message}")

    if job.state in self._BLOCKING_STATUSES:
        self._raise_for_blocking_status(job)

    self._raise_if_audience_cannot_produce_responses()

    # Get progress from pipeline if available
    with tqdm(
        total=100,
        desc="Processing job",
        unit="%",
        bar_format="{desc}: {percentage:3.0f}%|{bar}| completed [{elapsed}<{remaining}, {rate_fmt}]",
        disable=rapidata_config.logging.silent_mode,
    ) as pbar:
        last_percentage = 0
        while True:
            job = self._fetch_job()

            if job.state == AudienceJobState.COMPLETED:
                pbar.update(100 - last_percentage)
                break

            if job.state == AudienceJobState.FAILED:
                raise Exception(f"Job '{self}' has failed: {job.failure_message}")

            if job.state in self._BLOCKING_STATUSES:
                self._raise_for_blocking_status(job)

            # Try to get progress from workflow
            try:
                progress = self._get_workflow_progress()
                current_percentage = (
                    progress.completion_percentage if progress else 0
                )

                if current_percentage > last_percentage:
                    pbar.update(current_percentage - last_percentage)
                    last_percentage = current_percentage
            except Exception:
                pass  # Continue without progress update if we can't get it

            sleep(refresh_rate)

delete

delete() -> None

Deletes the job.

Source code in src/rapidata/rapidata_client/job/rapidata_job.py

def delete(self) -> None:
    """Deletes the job."""
    with tracer.start_as_current_span("RapidataJob.delete"):
        logger.info("Deleting job '%s'", self)
        self._openapi_service.order.job_api.job_job_id_delete(self.id)
        logger.debug("Job '%s' has been deleted.", self)
        managed_print(f"Job '{self}' has been deleted.")

view

view() -> None

Opens the job details page in the browser.

Source code in src/rapidata/rapidata_client/job/rapidata_job.py

def view(self) -> None:
    """Opens the job details page in the browser."""
    logger.info("Opening job details page in browser...")
    if not webbrowser.open(self.job_details_page):
        encoded_url = urllib.parse.quote(
            self.job_details_page, safe="%/:=&?~#+!$,;'@()*[]"
        )
        managed_print(
            Fore.RED
            + f"Please open this URL in your browser: '{encoded_url}'"
            + Fore.RESET
        )