Skip to content

Rapidata job definition

RapidataJobDefinition #

RapidataJobDefinition(
    id: str, name: str, openapi_service: OpenAPIService
)
Source code in src/rapidata/rapidata_client/job/rapidata_job_definition.py 
15
16
17
18
19
20
21
22
23
24
25
26
def __init__(
    self,
    id: str,
    name: str,
    openapi_service: OpenAPIService,
):
    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}"
    )

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 
28
29
30
31
32
33
34
35
36
37
38
39
40
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 
42
43
44
45
46
47
48
49
50
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.")

update_dataset #

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

Update the dataset of the job definition.

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'
Source code in src/rapidata/rapidata_client/job/rapidata_job_definition.py 
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
def update_dataset(
    self,
    datapoints: list[str] | list[list[str]],
    data_type: Literal["text", "media"] = "media",
    contexts: list[str] | None = None,
    media_contexts: list[str] | None = None,
    sentences: list[str] | None = None,
    private_metadata: list[dict[str, str]] | None = None,
) -> RapidataJobDefinition:
    """Update the dataset of the job definition.

    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.
    """
    with tracer.start_as_current_span("JobDefinition.update_dataset"):
        from rapidata.rapidata_client.datapoints._datapoints_validator import (
            DatapointsValidator,
        )
        from rapidata.api_client.models.create_dataset_endpoint_input import (
            CreateDatasetEndpointInput,
        )
        from rapidata.rapidata_client.dataset._rapidata_dataset import (
            RapidataDataset,
        )
        from rapidata.api_client.models.create_job_revision_endpoint_input import (
            CreateJobRevisionEndpointInput,
        )

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

        dataset = self._openapi_service.dataset.dataset_api.dataset_post(
            create_dataset_endpoint_input=CreateDatasetEndpointInput(
                name=self.name + "_dataset"
            )
        )

        rapidata_dataset = RapidataDataset(
            dataset.dataset_id, self._openapi_service
        )

        with tracer.start_as_current_span("update_datapoints"):
            _, failed_uploads = rapidata_dataset.add_datapoints(datapoints_list)
            if failed_uploads:
                raise FailedUploadException(
                    rapidata_dataset, failed_uploads, job_definition=self
                )

        self._openapi_service.order.job_api.job_definition_definition_id_revision_post(
            definition_id=self.id,
            create_job_revision_endpoint_input=CreateJobRevisionEndpointInput(
                datasetId=rapidata_dataset.id,
            ),
        )

        return self