Rapidata job definition

RevisionInfo `dataclass` #

RevisionInfo(
    revision: int,
    base_revision: int | None,
    state: RevisionState,
    created_at: datetime,
)

One revision of a job definition, as listed by RapidataJobDefinition.revisions().

RapidataJobDefinition #

RapidataJobDefinition(
    id: str,
    name: str,
    openapi_service: OpenAPIService,
    revision: int | None = None,
)

A job definition, held at one revision.

Reading a property returns the loaded revision's value, or your unsaved change. Assigning one stages the change locally; save() sends the whole revision in one request and moves the object to the new revision. audience.assign_job runs the revision this object holds.

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

def __init__(
    self,
    id: str,
    name: str,
    openapi_service: OpenAPIService,
    revision: int | None = None,
):
    self.id = id
    self.name = name
    self._openapi_service = openapi_service
    self._job_details_page = (
        f"https://app.{self._openapi_service.environment}/definitions/{self.id}"
    )
    self.__estimated_cost: CostEstimate | None = None
    self._revision_number = revision
    self._loaded_snapshot: GetJobRevisionEndpointOutput | None = None
    self._pending: dict[str, Any] = {}

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 definition 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.

revision `property` #

revision: int

The revision this object holds. assign_job runs exactly this revision.

base_revision `property` #

base_revision: int | None

The revision this one was edited from, or None if it was not recorded.

state `property` #

state: RevisionState

The current state of this revision, fetched from the server.

A new revision is Pending until its preview has been built.

dataset_id `property` #

dataset_id: str

The dataset of this revision, including an unsaved set_datapoints().

instruction `property` `writable` #

instruction: str | None

The instruction shown to labelers. None when it is set per datapoint.

answer_options `property` `writable` #

answer_options: list[str] | None

The answer options of a classification job definition.

None when the options are set per datapoint (the default for job definitions created with create_classification_job_definition) or the job type has none. Setting it shows the same options for every datapoint.

responses_per_datapoint `property` `writable` #

responses_per_datapoint: int

The maximum number of responses collected per datapoint.

settings `property` `writable` #

settings: list[RapidataSetting]

The settings of this revision, as CustomSetting objects.

Assigning a list replaces every setting, at both the rapid and campaign level.

validation_set_id `property` `writable` #

validation_set_id: str | None

The validation set pinned on this revision, or None for none.

has_unsaved_changes `property` #

has_unsaved_changes: bool

Whether a change is staged that save() has not sent yet.

preview #

preview() -> RapidataJobDefinition

Will open the browser where you can preview the job definition before giving it to an audience.

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

def preview(self) -> RapidataJobDefinition:
    """Will open the browser where you can preview the job definition before giving it to an audience."""
    logger.info("Opening order 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
        )
    return self

delete #

delete() -> None

Deletes the job definition and all its revisions.

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

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

set_datapoints #

set_datapoints(
    datapoints: list[str] | list[list[str]],
    data_type: Literal["text", "media"] = "media",
    contexts: list[str] | None = None,
    media_contexts: list[list[str]] | None = None,
    sentences: list[str] | None = None,
    private_metadata: list[dict[str, str]] | None = None,
    answer_options: (
        list[str] | list[list[str]] | None
    ) = None,
) -> None

Upload a new dataset now and stage it for the next save().

Parameters:

Name	Type	Description	Default
`datapoints`	`list[str] \| list[list[str]]`	paths to the datapoints or strings for text datapoints.	required
`data_type`	`Literal['text', 'media']`	The type of the datapoints.	`'media'`
`contexts`	`list[str]`	One text context per datapoint. Defaults to None.	`None`
`media_contexts`	`list[list[str]]`	Image URLs / paths shown alongside each datapoint, one inner list per datapoint. Defaults to None.	`None`
`sentences`	`list[str]`	The sentences associated with each datapoint (used by select-words workflows). Defaults to None.	`None`
`private_metadata`	`list[dict[str, str]]`	Key-value pairs attached to each datapoint that will not be shown to labelers. Defaults to None.	`None`
`answer_options`	`list[str] \| list[list[str]]`	Classification only, required when the options are set per datapoint: one list for every datapoint, or one list per datapoint.	`None`

Raises:

Type	Description
`FailedUploadException`	If any datapoint fails to upload. Nothing is staged.

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

def set_datapoints(
    self,
    datapoints: list[str] | list[list[str]],
    data_type: Literal["text", "media"] = "media",
    contexts: list[str] | None = None,
    media_contexts: list[list[str]] | None = None,
    sentences: list[str] | None = None,
    private_metadata: list[dict[str, str]] | None = None,
    answer_options: list[str] | list[list[str]] | None = None,
) -> None:
    """Upload a new dataset now and stage it for the next ``save()``.

    Args:
        datapoints (list[str] | list[list[str]]): paths to the datapoints or strings for text datapoints.
        data_type (Literal["text", "media"]): The type of the datapoints.
        contexts (list[str], optional): One text context per datapoint. Defaults to None.
        media_contexts (list[list[str]], optional): Image URLs / paths shown alongside each datapoint, one inner list per datapoint. Defaults to None.
        sentences (list[str], optional): The sentences associated with each datapoint (used by select-words workflows). Defaults to None.
        private_metadata (list[dict[str, str]], optional): Key-value pairs attached to each datapoint that will not be shown to labelers. Defaults to None.
        answer_options (list[str] | list[list[str]], optional): Classification only, required when the
            options are set per datapoint: one list for every datapoint, or one list per datapoint.

    Raises:
        FailedUploadException: If any datapoint fails to upload. Nothing is staged.
    """
    with tracer.start_as_current_span("RapidataJobDefinition.set_datapoints"):
        from rapidata.rapidata_client.datapoints._datapoints_validator import (
            DatapointsValidator,
        )
        from rapidata.rapidata_client.datapoints._default_segments import (
            ANSWER_OPTIONS_SEGMENT_KEY,
            build_schema,
        )
        from rapidata.api_client.models.create_dataset_endpoint_input import (
            CreateDatasetEndpointInput,
        )
        from rapidata.rapidata_client.dataset._rapidata_dataset import (
            RapidataDataset,
        )
        from rapidata.rapidata_client.job._revision_body import (
            _required_segments,
        )

        workflow = self._snapshot.workflow
        if "answer_options" in self._pending:
            workflow = _with_answer_options(
                workflow, self._pending["answer_options"]
            )
        required = _required_segments(workflow)

        per_datapoint_options: list[list[str]] | None = None
        if answer_options is not None:
            if all(isinstance(o, list) for o in answer_options):
                per_datapoint_options = [list(o) for o in answer_options]  # type: ignore[arg-type]
            elif all(isinstance(o, str) for o in answer_options):
                per_datapoint_options = [list(answer_options)] * len(datapoints)  # type: ignore[arg-type]
            else:
                raise ValueError(
                    "answer_options must be either a list of strings or a list of lists of strings"
                )
            required.add(ANSWER_OPTIONS_SEGMENT_KEY)
        elif ANSWER_OPTIONS_SEGMENT_KEY in required:
            raise ValueError(
                "This job definition reads its answer options from each datapoint; "
                "pass answer_options to set_datapoints()."
            )

        datapoints_list = DatapointsValidator.map_datapoints(
            datapoints=datapoints,
            contexts=contexts,
            media_contexts=media_contexts,
            sentences=sentences,
            private_metadata=private_metadata,
            data_type=data_type,
            answer_options=per_datapoint_options,
        )

        dataset = self._openapi_service.dataset.dataset_api.dataset_post(
            create_dataset_endpoint_input=CreateDatasetEndpointInput(
                name=self.name + "_dataset",
                schema=build_schema(required) if required else None,
            )
        )
        rapidata_dataset = RapidataDataset(
            dataset.dataset_id, self._openapi_service
        )

        _, failed_uploads = rapidata_dataset.add_datapoints(datapoints_list)
        if failed_uploads:
            raise FailedUploadException(
                rapidata_dataset, failed_uploads, job_definition=self
            )

        self._pending["dataset_id"] = rapidata_dataset.id

save #

save(require_latest: bool = False) -> int

Create a new revision from this object and move the object to it.

The request carries the complete revision: the one this object was loaded at, plus your staged changes. Changes saved by someone else since you loaded are not merged in; with require_latest=False (default) your revision becomes the latest anyway.

Parameters:

Name	Type	Description	Default
`require_latest`	`bool`	Only save if no other revision was saved since this object was loaded. Defaults to False.	`False`

Returns:

Name	Type	Description
`int`	`int`	The new revision number.

Raises:

Type	Description
`RevisionConflictError`	With `require_latest=True`, when another revision was saved in the meantime. Nothing is created and your changes stay staged.

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

def save(self, require_latest: bool = False) -> int:
    """Create a new revision from this object and move the object to it.

    The request carries the complete revision: the one this object was loaded at,
    plus your staged changes. Changes saved by someone else since you loaded are
    not merged in; with ``require_latest=False`` (default) your revision becomes
    the latest anyway.

    Args:
        require_latest (bool): Only save if no other revision was saved since this
            object was loaded. Defaults to False.

    Returns:
        int: The new revision number.

    Raises:
        RevisionConflictError: With ``require_latest=True``, when another revision
            was saved in the meantime. Nothing is created and your changes stay staged.
    """
    with tracer.start_as_current_span("RapidataJobDefinition.save"):
        from rapidata.rapidata_client.api.rapidata_api_client import (
            suppress_rapidata_error_logging,
        )
        from rapidata.rapidata_client.exceptions.rapidata_error import (
            RapidataError,
        )

        snapshot = self._snapshot
        body = _full_body(snapshot, self._pending, require_latest)
        try:
            with suppress_rapidata_error_logging():
                response = self._openapi_service.order.job_api.job_definition_definition_id_revision_post(
                    definition_id=self.id,
                    create_job_revision_endpoint_input=body,
                )
        except RapidataError as e:
            if e.status_code == 409 and isinstance(e.details, dict):
                raise RevisionConflictError(
                    job_definition_id=self.id,
                    base_revision=snapshot.revision_number,
                    latest_revision=e.details.get("latestRevisionNumber", -1),
                ) from None
            logger.error("Error: %s", e)
            raise

        self._revision_number = response.revision_number
        self._loaded_snapshot = None
        self._pending = {}
        logger.info(
            "Saved job definition '%s' as revision %s (from %s)",
            self.id,
            response.revision_number,
            snapshot.revision_number,
        )
        return response.revision_number

reload #

reload(
    revision: int | None = None,
) -> RapidataJobDefinition

Load a revision into this object, the latest one by default.

Raises:

Type	Description
`RuntimeError`	If there are unsaved changes; `save()` them or `discard_changes()` first.

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

def reload(self, revision: int | None = None) -> RapidataJobDefinition:
    """Load a revision into this object, the latest one by default.

    Raises:
        RuntimeError: If there are unsaved changes; ``save()`` them or ``discard_changes()`` first.
    """
    if self._pending:
        raise RuntimeError(
            f"Job definition '{self.id}' has unsaved changes to "
            f"{', '.join(sorted(self._pending))}. Call save() or discard_changes() first."
        )
    with tracer.start_as_current_span("RapidataJobDefinition.reload"):
        self._loaded_snapshot = self._fetch_revision(revision)
        self._revision_number = self._loaded_snapshot.revision_number
        return self

discard_changes #

discard_changes() -> RapidataJobDefinition

Drop every staged change; the object keeps its loaded revision.

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

def discard_changes(self) -> RapidataJobDefinition:
    """Drop every staged change; the object keeps its loaded revision."""
    self._pending = {}
    return self

revisions #

revisions() -> list[RevisionInfo]

Every revision of this job definition, oldest first.

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

def revisions(self) -> list[RevisionInfo]:
    """Every revision of this job definition, oldest first."""
    with tracer.start_as_current_span("RapidataJobDefinition.revisions"):
        job_api = self._openapi_service.order.job_api
        infos: list[RevisionInfo] = []
        page = 1
        while True:
            result = job_api.job_definition_definition_id_revisions_get(
                definition_id=self.id,
                page=page,
                page_size=100,
                sort=["revision_number"],
            )
            infos.extend(
                RevisionInfo(
                    revision=item.revision_number,
                    base_revision=item.base_revision_number,
                    state=item.state.value,  # type: ignore[arg-type]
                    created_at=item.created_at,
                )
                for item in result.items
            )
            if len(result.items) < 100:
                return infos
            page += 1

wait_until_ready #

wait_until_ready(
    timeout: float = 300,
) -> RapidataJobDefinition

Block until this revision's preview is built.

Raises:

Type	Description
`TimeoutError`	If it is still `Pending` after `timeout` seconds.
`RuntimeError`	If the revision failed.

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

def wait_until_ready(self, timeout: float = 300) -> RapidataJobDefinition:
    """Block until this revision's preview is built.

    Raises:
        TimeoutError: If it is still ``Pending`` after ``timeout`` seconds.
        RuntimeError: If the revision failed.
    """
    with tracer.start_as_current_span("RapidataJobDefinition.wait_until_ready"):
        deadline = time.monotonic() + timeout
        while (state := self.state) == "Pending":
            if time.monotonic() >= deadline:
                raise TimeoutError(
                    f"Revision {self.revision} of job definition '{self.id}' "
                    f"is still Pending after {timeout} seconds."
                )
            time.sleep(2)
        if state == "Failed":
            raise RuntimeError(
                f"Revision {self.revision} of job definition '{self.id}' failed."
            )
        return self

update_dataset #

update_dataset(
    datapoints: list[str] | list[list[str]],
    data_type: Literal["text", "media"] = "media",
    contexts: list[str] | None = None,
    media_contexts: list[list[str]] | None = None,
    sentences: list[str] | None = None,
    private_metadata: list[dict[str, str]] | None = None,
) -> RapidataJobDefinition

Deprecated: use set_datapoints(...) followed by save().

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

def update_dataset(
    self,
    datapoints: list[str] | list[list[str]],
    data_type: Literal["text", "media"] = "media",
    contexts: list[str] | None = None,
    media_contexts: list[list[str]] | None = None,
    sentences: list[str] | None = None,
    private_metadata: list[dict[str, str]] | None = None,
) -> RapidataJobDefinition:
    """Deprecated: use ``set_datapoints(...)`` followed by ``save()``."""
    warnings.warn(
        "update_dataset() is deprecated; call set_datapoints(...) and then save().",
        DeprecationWarning,
        stacklevel=2,
    )
    self.set_datapoints(
        datapoints=datapoints,
        data_type=data_type,
        contexts=contexts,
        media_contexts=media_contexts,
        sentences=sentences,
        private_metadata=private_metadata,
    )
    self.save()
    return self

Rapidata job definition

RevisionInfo dataclass #