diff --git a/.env.example b/.env.example index f69c87bf7..40fab18a2 100644 --- a/.env.example +++ b/.env.example @@ -10,6 +10,7 @@ # RAPIDATA_cacheShards=128 # RAPIDATA_batchSize=1000 # RAPIDATA_batchPollInterval=0.5 +# RAPIDATA_failureTolerance=0.0 # --- Logging --- # RAPIDATA_level=WARNING diff --git a/docs/error_handling.md b/docs/error_handling.md index fed56a088..eac529746 100644 --- a/docs/error_handling.md +++ b/docs/error_handling.md @@ -2,38 +2,53 @@ ## Introduction -When creating job definitions or orders with the Rapidata SDK, datapoints may fail to upload due to various reasons such as missing files, invalid formats, or network issues. Understanding how to handle these failures is essential for building robust integrations. +When creating job definitions with the Rapidata SDK, datapoints may fail to upload for various reasons such as missing files, invalid formats, or network issues. Understanding how to handle these failures is essential for building robust integrations. -When one or more datapoints fail to upload, the SDK raises a `FailedUploadException`. This exception provides detailed information about what went wrong and gives you several recovery options: +Job creation is **atomic by default**: if any datapoint fails to upload, the SDK does **not** create the job definition — so you never end up with an incomplete definition — and raises a `FailedUploadException`. You then fix the problem and call `exception.retry()`, which re-uploads only the failed datapoints into the **same** dataset and finishes creating the definition. Retrying never creates a duplicate dataset or definition. -- Inspect which datapoints failed and why -- Retry the failed datapoints -- Continue with the successfully uploaded datapoints +If you would rather tolerate a small fraction of failures instead, raise the [failure tolerance](#failure-tolerance). -This guide shows you how to handle upload failures effectively. +## Failure tolerance -## Understanding FailedUploadException +`failure_tolerance` is the fraction of a job's datapoints (`0.0`–`1.0`) allowed to fail while still creating the definition: -The `FailedUploadException` is raised during `JobDefinition` or `Order` creation when one or more datapoints cannot be uploaded. -**Important**: Despite the exception being raised, a `JobDefinition` or `Order` object is still created with the successfully uploaded datapoints, allowing you to continue if you catch the exception. +- `0.0` (default) — strict. Any failed upload aborts creation; no definition is left behind and the failed datapoints can be retried. +- `0.01` — up to 1% of datapoints may fail; the definition is created with the rest, and the failures are logged. +- `1.0` — the definition is created regardless of how many datapoints failed, as long as at least one datapoint uploads successfully (a definition over an empty dataset is never created). -### Exception Properties +Set it globally, via the config object or the `RAPIDATA_failureTolerance` environment variable: + +```python +from rapidata import rapidata_config +rapidata_config.upload.failureTolerance = 0.01 +``` -The exception provides these properties to help you understand and recover from failures: +or per call, which overrides the global default for that job: ```python -FailedUploadException( - dataset: RapidataDataset, # (1)! - failed_uploads: list[FailedUpload], # (2)! - order: Optional[RapidataOrder], # (3)! - job_definition: Optional[JobDefinition] # (4)! +job_def = client.job.create_classification_job_definition( + name="Image Classification", + instruction="What animal is in this image?", + answer_options=["Cat", "Dog", "Bird"], + datapoints=["cat1.jpg", "dog1.jpg", "missing.jpg"], + failure_tolerance=0.01, ) ``` -1. The dataset that was being created. -2. Basic list of failed datapoints. -3. The order object (only present during order creation). -4. The job definition object (only present during job definition creation). +The ratio is measured against the whole job, so it stays meaningful across retries. + +## Understanding FailedUploadException + +The `FailedUploadException` is raised during job-definition creation when the upload stays outside the failure tolerance. When it comes from job creation, no job definition was created — recover with `exception.retry()` (see below). + +### Exception Properties + +The exception exposes these to help you understand and recover from failures: + +- `dataset` — the dataset that was being uploaded to. `retry()` reuses it. +- `failed_uploads` / `detailed_failures` / `failures_by_reason` — which datapoints failed and why (see below). +- `retry()` — re-upload the failed datapoints into `dataset` and finish creating the definition. Returns the `RapidataJobDefinition`. +- `job_definition` — `None` for job creation (nothing was persisted). Populated only when tolerating failures on a legacy order. ### Understanding Failure Information @@ -99,11 +114,11 @@ This groups all failed datapoints by their error message, making it easy to see **Datapoint Creation Failures**: After assets are successfully uploaded, datapoints are created. These failures can have different reasons depending on what went wrong (e.g., validation errors, format issues, backend constraints). Each datapoint may fail for a unique reason. -## Recovery Strategies +## Recovery -### Strategy 1: Continue with Successfully Uploaded Datapoints +### Fix and retry -When a `FailedUploadException` is raised, the `JobDefinition` is still created with the successfully uploaded datapoints. You can catch the exception and continue using the created object: +The recommended recovery path is to catch the exception, fix whatever caused the failures (for example correct the file paths), and call `exception.retry()`. This re-uploads **only** the failed datapoints into the **same** dataset and finishes creating the definition — no new dataset, no duplicate definition, and nothing to re-specify: ```python from rapidata import RapidataClient @@ -116,67 +131,46 @@ try: name="Image Classification", instruction="What animal is in this image?", answer_options=["Cat", "Dog", "Bird"], - datapoints=["cat1.jpg", "dog1.jpg", "missing.jpg"] + datapoints=["cat1.jpg", "dog1.jpg", "missing.jpg"], ) except FailedUploadException as e: - print(f"Warning: {len(e.failed_uploads)} datapoints failed to upload") - - if len(e.failed_uploads) > len(datapoints) * 0.1: # (1)! - raise ValueError("Too many failures, aborting") + for reason, datapoints in e.failures_by_reason.items(): # (1)! + print(f" {reason}: {len(datapoints)} datapoints") - job_def = e.job_definition # (2)! + # ...fix the failing datapoints (e.g. correct the paths on disk)... + job_def = e.retry() # (2)! -# The job definition holds the successful datapoints — assign it to an audience to start collecting responses. audience = client.audience.get_audience_by_id("global") job = audience.assign_job(job_def) ``` -1. Check if the failure rate is acceptable — here we abort if more than 10% failed. -2. The job definition was still created with the successfully uploaded datapoints. You can use it normally. - -### Strategy 2: Retry Failed Datapoints - -After catching the exception, you can fix the issues (e.g., correct file paths, fix formats) and retry the failed datapoints by adding them to the dataset: - -```python -from rapidata import RapidataClient -from rapidata.rapidata_client.exceptions import FailedUploadException +1. Inspect what failed and why before retrying. +2. Returns the created `RapidataJobDefinition`. If some datapoints still fail beyond the tolerance, `retry()` raises `FailedUploadException` again (with the same dataset attached), so you can keep fixing and retrying. -client = RapidataClient() +!!! warning + Don't re-call `create_*_job_definition(...)` to retry — that starts a fresh dataset and creates a **second** definition. Use `exception.retry()` so the existing dataset is reused. -try: - job_def = client.job.create_classification_job_definition( - name="Image Classification", - instruction="What animal is in this image?", - answer_options=["Cat", "Dog", "Bird"], - datapoints=["cat1.jpg", "dog1.jpg", "missing.jpg"] - ) -except FailedUploadException as e: - print(f"{len(e.failed_uploads)} datapoints failed:") - for reason, datapoints in e.failures_by_reason.items(): - print(f" {reason}: {len(datapoints)} datapoints") +### Tolerate a fraction of failures - successful_retries, failed_retries = e.dataset.add_datapoints(e.failed_uploads) # (1)! - print(f"{len(successful_retries)} datapoints successfully added on retry") +If a few failures are acceptable, set a [failure tolerance](#failure-tolerance). When the failed fraction is within tolerance the definition is created with the datapoints that succeeded, the failures are logged, and no exception is raised: - if failed_retries: - print(f"{len(failed_retries)} datapoints still failed after retry") +```python +job_def = client.job.create_classification_job_definition( + name="Image Classification", + instruction="What animal is in this image?", + answer_options=["Cat", "Dog", "Bird"], + datapoints=["cat1.jpg", "dog1.jpg", "missing.jpg"], + failure_tolerance=0.5, # tolerate up to 50% failures for this job +) ``` -1. Fix the underlying issues (e.g., correct file paths) before retrying. This adds the previously failed datapoints back to the dataset. - -### Strategy 3: Retrieve and Use After Exception (If Not Caught) +### Manual control (advanced) -If you didn't catch the exception during creation, you can still retrieve and use the job definition. It was created with the successfully uploaded datapoints and can be used through code or the app.rapidata.ai UI: +`exception.retry()` covers the common case. If you need to drive the retry yourself — for example to substitute corrected datapoints — the failed dataset is available on the exception and you can add datapoints to it directly: ```python -from rapidata import RapidataClient - -client = RapidataClient() - -job_def = client.job.get_job_definition_by_id(job_definition_id) # (1)! -audience = client.audience.get_audience_by_id("global") -audience.assign_job(job_def) +except FailedUploadException as e: + successful_retries, failed_retries = e.dataset.add_datapoints(e.failed_uploads) ``` -1. Retrieve the job definition using its ID (from the exception message or the [Rapidata Dashboard](https://app.rapidata.ai)). +Note that this only re-uploads the datapoints; it does not create the job definition. Prefer `exception.retry()` unless you specifically need the lower-level control. diff --git a/src/rapidata/rapidata_client/config/upload_config.py b/src/rapidata/rapidata_client/config/upload_config.py index 900c447a3..ec3849a37 100644 --- a/src/rapidata/rapidata_client/config/upload_config.py +++ b/src/rapidata/rapidata_client/config/upload_config.py @@ -103,6 +103,14 @@ class UploadConfig(BaseModel): maximum length is automatically shortened for the order/job instruction before upload. When False (default), an over-long context is left unchanged and a warning is logged that the backend would reject it. Defaults to False. + failureTolerance (float): The fraction of a job's datapoints allowed to fail while + still creating the job definition (0.0-1.0). 0.0 (default) is strict: any failed + upload aborts creation so no incomplete definition is left behind, and the failed + datapoints can be retried into the same dataset. 1.0 creates the definition + regardless of how many datapoints failed, as long as at least one datapoint + uploads successfully (a definition over an empty dataset is never created). + Overridable per call via the ``failure_tolerance`` argument on + ``create_*_job_definition``. Defaults to 0.0. """ model_config = ConfigDict(validate_assignment=True) @@ -145,6 +153,10 @@ def _apply_env_vars(cls, data: Any) -> Any: default=False, description="Automatically shorten over-long datapoint contexts for the instruction before upload.", ) + failureTolerance: float = Field( + default=0.0, + description="Fraction of a job's datapoints allowed to fail while still creating the definition (0.0-1.0).", + ) @field_validator("maxWorkers") @classmethod @@ -174,6 +186,13 @@ def validate_batch_size(cls, v: int) -> int: raise ValueError("batchSize must be at least 100") return v + @field_validator("failureTolerance") + @classmethod + def validate_failure_tolerance(cls, v: float) -> float: + if not 0.0 <= v <= 1.0: + raise ValueError("failureTolerance must be between 0.0 and 1.0") + return v + def __init__(self, **kwargs): super().__init__(**kwargs) self._migrate_cache() diff --git a/src/rapidata/rapidata_client/exceptions/failed_upload_exception.py b/src/rapidata/rapidata_client/exceptions/failed_upload_exception.py index 6da1456e7..566e96a86 100644 --- a/src/rapidata/rapidata_client/exceptions/failed_upload_exception.py +++ b/src/rapidata/rapidata_client/exceptions/failed_upload_exception.py @@ -9,6 +9,9 @@ from rapidata.rapidata_client.job.rapidata_job_definition import ( RapidataJobDefinition, ) + from rapidata.rapidata_client.job._job_creation_state_machine import ( + JobDefinitionCreationMachine, + ) from rapidata.rapidata_client.order.rapidata_order import RapidataOrder @@ -21,13 +24,44 @@ def __init__( failed_uploads: list[FailedUpload[Datapoint]], order: Optional[RapidataOrder] = None, job_definition: Optional[RapidataJobDefinition] = None, + machine: Optional[JobDefinitionCreationMachine] = None, ): self.dataset = dataset self.order = order self.job_definition = job_definition self._failed_uploads = failed_uploads + self._machine = machine super().__init__(str(self)) + def retry(self) -> RapidataJobDefinition: + """Retry the failed datapoints and finish creating the job definition. + + Re-uploads only the datapoints that failed into the **same** dataset + (never a new one), then creates the job definition once the upload is + within the configured failure tolerance. Returns the created + ``RapidataJobDefinition``. If some datapoints still fail beyond the + tolerance, this raises ``FailedUploadException`` again (with the same + dataset attached) so it can be caught and retried in a loop. + + This is the intended recovery path for a failed job-definition + creation: fix whatever caused the failures (e.g. correct file paths), + then call ``retry()`` - no need to re-specify the datapoints or drop + down to ``dataset.add_datapoints`` manually. + """ + if self._machine is None: + raise RuntimeError( + "retry() is only available for failed job-definition creation. " + "To re-upload the failed datapoints manually, use " + "dataset.add_datapoints(exception.failed_uploads)." + ) + return self._machine.resume() + + @property + def machine(self) -> Optional[JobDefinitionCreationMachine]: + """The creation state machine backing ``retry()``, when this failure + came from job-definition creation (``None`` for order uploads).""" + return self._machine + @property def failed_uploads(self) -> list[Datapoint]: """ @@ -85,6 +119,13 @@ def __str__(self) -> str: lines.append(" ]") failed_upload_message = "\n".join(lines) + if self.machine is not None: + failed_upload_message += ( + "\n\nNo job definition was created because the upload stayed outside the " + "failure tolerance. Fix the failed datapoints and retry them into the same " + "dataset (no new dataset is created) by catching this exception and calling: " + "\n\tjob_definition = exception.retry()" + ) if self.order: failed_upload_message += f"\n\nTo run the order without the failed datapoints, call: \n\trapidata_client.order.get_order_by_id('{self.order.id}').run()" if self.job_definition: diff --git a/src/rapidata/rapidata_client/job/_job_creation_state_machine.py b/src/rapidata/rapidata_client/job/_job_creation_state_machine.py new file mode 100644 index 000000000..8cedae108 --- /dev/null +++ b/src/rapidata/rapidata_client/job/_job_creation_state_machine.py @@ -0,0 +1,208 @@ +from __future__ import annotations + +from enum import Enum, auto +from typing import TYPE_CHECKING, Sequence + +from rapidata.rapidata_client.config import logger, tracer +from rapidata.rapidata_client.config._qr_preview import ( + print_campaign_preview_qr_for_pipeline, + print_job_definition_preview_link, +) +from rapidata.rapidata_client.dataset._rapidata_dataset import RapidataDataset +from rapidata.rapidata_client.exceptions.failed_upload import FailedUpload +from rapidata.rapidata_client.exceptions.failed_upload_exception import ( + FailedUploadException, +) +from rapidata.rapidata_client.job.rapidata_job_definition import RapidataJobDefinition + +if TYPE_CHECKING: + from rapidata.api_client.models.feature_flag import FeatureFlag + from rapidata.rapidata_client.datapoints._datapoint import Datapoint + from rapidata.rapidata_client.referee._base_referee import Referee + from rapidata.rapidata_client.workflow import Workflow + from rapidata.service.openapi_service import OpenAPIService + + +class _State(Enum): + CREATE_DATASET = auto() + UPLOAD_DATAPOINTS = auto() + PERSIST_DEFINITION = auto() + SUCCEEDED = auto() + FAILED = auto() + + +class JobDefinitionCreationMachine: + """Drives job-definition creation as an explicit state machine. + + The remote job definition is persisted only once the datapoint upload has + landed within the configured failure tolerance, so a partially-uploaded + dataset never leaves behind an incomplete definition. The machine keeps all + intermediate state (the dataset, the datapoints still pending, the upload + failures) so a failed run can be resumed via :meth:`resume` — re-uploading + only the datapoints that failed into the *same* dataset rather than starting + a fresh one. This is what backs ``FailedUploadException.retry()``. + + ``failure_tolerance`` is the fraction of the job's datapoints allowed to fail + while still creating the definition (``0.0`` = strict, ``1.0`` = create + regardless). The ratio is always measured against the original datapoint + count, so it stays meaningful across resume attempts. Regardless of the + tolerance, at least one datapoint must upload successfully - a definition + over an empty dataset is never created. + """ + + def __init__( + self, + openapi_service: OpenAPIService, + name: str, + workflow: Workflow, + datapoints: list[Datapoint], + referee: Referee, + failure_tolerance: float, + rapid_feature_flags: Sequence[FeatureFlag] | None = None, + campaign_feature_flags: Sequence[FeatureFlag] | None = None, + ): + self._openapi_service = openapi_service + self._name = name + self._workflow = workflow + self._referee = referee + self._failure_tolerance = failure_tolerance + self._rapid_feature_flags = rapid_feature_flags + self._campaign_feature_flags = campaign_feature_flags + + self._total_datapoints = len(datapoints) + self._pending: list[Datapoint] = list(datapoints) + self._succeeded_count = 0 + + self._state = _State.CREATE_DATASET + self.dataset: RapidataDataset | None = None + self.failed_uploads: list[FailedUpload[Datapoint]] = [] + self.job_definition: RapidataJobDefinition | None = None + + def run(self) -> RapidataJobDefinition: + """Drive the machine to a terminal state. + + Returns the created definition on success, or raises + ``FailedUploadException`` (with this machine attached, so the caller can + ``retry()``) when the upload stays outside the failure tolerance. + """ + while self._state not in (_State.SUCCEEDED, _State.FAILED): + if self._state is _State.CREATE_DATASET: + self._create_dataset() + elif self._state is _State.UPLOAD_DATAPOINTS: + self._upload_datapoints() + elif self._state is _State.PERSIST_DEFINITION: + self._persist_definition() + + if self._state is _State.FAILED: + # FAILED is only ever entered from _upload_datapoints, which runs + # after the dataset exists. + assert self.dataset is not None + raise FailedUploadException( + self.dataset, + self.failed_uploads, + job_definition=None, + machine=self, + ) + + assert self.job_definition is not None + return self.job_definition + + def resume(self) -> RapidataJobDefinition: + """Retry a failed run, reusing the existing dataset. + + Only the datapoints that previously failed are re-uploaded (into the + same dataset), then the machine re-evaluates the failure tolerance and + persists the definition if it now fits. + """ + # resume() is only reached via a raised FailedUploadException, which is + # only produced once the dataset exists. + assert self.dataset is not None + self._pending = [fu.item for fu in self.failed_uploads] + self.failed_uploads = [] + self._state = _State.UPLOAD_DATAPOINTS + return self.run() + + def _create_dataset(self) -> None: + from rapidata.api_client.models.create_dataset_endpoint_input import ( + CreateDatasetEndpointInput, + ) + + dataset = self._openapi_service.dataset.dataset_api.dataset_post( + create_dataset_endpoint_input=CreateDatasetEndpointInput( + name=self._name + "_dataset" + ) + ) + self.dataset = RapidataDataset(dataset.dataset_id, self._openapi_service) + self._state = _State.UPLOAD_DATAPOINTS + + def _upload_datapoints(self) -> None: + assert self.dataset is not None + with tracer.start_as_current_span("add_datapoints"): + successful, failed = self.dataset.add_datapoints(self._pending) + + self._succeeded_count += len(successful) + self.failed_uploads = failed + self._pending = [] + + failed_count = self._total_datapoints - self._succeeded_count + failure_ratio = ( + failed_count / self._total_datapoints if self._total_datapoints else 0.0 + ) + + # At least one datapoint must land regardless of the tolerance: a + # definition over an empty dataset has nothing to label and would be + # exactly the kind of useless remote artifact this machine prevents. + within_tolerance = failure_ratio <= self._failure_tolerance + if within_tolerance and self._succeeded_count > 0: + if failed_count > 0: + logger.warning( + "%d/%d datapoint(s) failed to upload (%.2f%%), which is within the " + "failure tolerance of %.2f%% - creating the job definition without them.", + failed_count, + self._total_datapoints, + failure_ratio * 100, + self._failure_tolerance * 100, + ) + self._state = _State.PERSIST_DEFINITION + else: + self._state = _State.FAILED + + def _persist_definition(self) -> None: + assert self.dataset is not None + from rapidata.api_client.models.create_job_definition_endpoint_input import ( + CreateJobDefinitionEndpointInput, + ) + + response = self._openapi_service.order.job_api.job_definition_post( + create_job_definition_endpoint_input=CreateJobDefinitionEndpointInput( + definitionName=self._name, + workflow=self._workflow._to_model(), + datasetId=self.dataset.id, + referee=self._referee._to_model(), + rapidFeatureFlags=( + list(self._rapid_feature_flags) + if self._rapid_feature_flags + else None + ), + campaignFeatureFlags=( + list(self._campaign_feature_flags) + if self._campaign_feature_flags + else None + ), + ) + ) + self.job_definition = RapidataJobDefinition( + id=response.definition_id, + name=self._name, + openapi_service=self._openapi_service, + ) + + print_campaign_preview_qr_for_pipeline( + openapi_service=self._openapi_service, + pipeline_id=response.pipeline_id, + ) + print_job_definition_preview_link( + environment=self._openapi_service.environment, + job_definition_id=self.job_definition.id, + ) + self._state = _State.SUCCEEDED diff --git a/src/rapidata/rapidata_client/job/rapidata_job_manager.py b/src/rapidata/rapidata_client/job/rapidata_job_manager.py index c0f9ab132..0fe2ae1d3 100644 --- a/src/rapidata/rapidata_client/job/rapidata_job_manager.py +++ b/src/rapidata/rapidata_client/job/rapidata_job_manager.py @@ -1,20 +1,15 @@ from __future__ import annotations from rapidata.service.openapi_service import OpenAPIService -from rapidata.rapidata_client.config import logger, tracer -from rapidata.rapidata_client.config._qr_preview import ( - print_campaign_preview_qr_for_pipeline, - print_job_definition_preview_link, -) +from rapidata.rapidata_client.config import logger, tracer, rapidata_config from rapidata.rapidata_client.datapoints._datapoint import Datapoint from rapidata.rapidata_client.workflow import Workflow from rapidata.rapidata_client.settings import RapidataSetting from rapidata.rapidata_client.job.rapidata_job_definition import RapidataJobDefinition -from typing import Sequence, Literal, TYPE_CHECKING -from rapidata.rapidata_client.dataset._rapidata_dataset import RapidataDataset -from rapidata.rapidata_client.exceptions.failed_upload_exception import ( - FailedUploadException, +from rapidata.rapidata_client.job._job_creation_state_machine import ( + JobDefinitionCreationMachine, ) +from typing import Sequence, Literal, TYPE_CHECKING from rapidata.rapidata_client.datapoints._datapoints_validator import ( DatapointsValidator, ) @@ -74,10 +69,14 @@ def _create_general_job_definition( confidence_threshold: float | None = None, quorum_threshold: int | None = None, settings: Sequence[RapidataSetting] | None = None, + failure_tolerance: float | None = None, ) -> RapidataJobDefinition: if settings is None: settings = [] + if failure_tolerance is not None and not 0.0 <= failure_tolerance <= 1.0: + raise ValueError("failure_tolerance must be between 0.0 and 1.0") + self._warn_unsupported_settings(workflow, settings) self.__context_manager._enforce_context_length( @@ -122,23 +121,6 @@ def _create_general_job_definition( quorum_threshold, settings, ) - from rapidata.api_client.models.create_dataset_endpoint_input import ( - CreateDatasetEndpointInput, - ) - from rapidata.api_client.models.create_job_definition_endpoint_input import ( - CreateJobDefinitionEndpointInput, - ) - - dataset = self._openapi_service.dataset.dataset_api.dataset_post( - create_dataset_endpoint_input=CreateDatasetEndpointInput( - name=name + "_dataset" - ) - ) - rapidata_dataset = RapidataDataset(dataset.dataset_id, self._openapi_service) - - with tracer.start_as_current_span("add_datapoints"): - _, failed_uploads = rapidata_dataset.add_datapoints(datapoints) - rapid_feature_flags = ( [s._to_feature_flag() for s in settings if s.target == "rapids"] if settings @@ -150,43 +132,23 @@ def _create_general_job_definition( else None ) - job_definition_response = ( - self._openapi_service.order.job_api.job_definition_post( - create_job_definition_endpoint_input=CreateJobDefinitionEndpointInput( - definitionName=name, - workflow=workflow._to_model(), - datasetId=rapidata_dataset.id, - referee=referee._to_model(), - rapidFeatureFlags=( - rapid_feature_flags if rapid_feature_flags else None - ), - campaignFeatureFlags=( - campaign_feature_flags if campaign_feature_flags else None - ), - ) - ) - ) - job_model = RapidataJobDefinition( - id=job_definition_response.definition_id, - name=name, - openapi_service=self._openapi_service, + tolerance = ( + failure_tolerance + if failure_tolerance is not None + else rapidata_config.upload.failureTolerance ) - if failed_uploads: - raise FailedUploadException( - rapidata_dataset, failed_uploads, job_definition=job_model - ) - - print_campaign_preview_qr_for_pipeline( + machine = JobDefinitionCreationMachine( openapi_service=self._openapi_service, - pipeline_id=job_definition_response.pipeline_id, - ) - print_job_definition_preview_link( - environment=self._openapi_service.environment, - job_definition_id=job_model.id, + name=name, + workflow=workflow, + datapoints=datapoints, + referee=referee, + failure_tolerance=tolerance, + rapid_feature_flags=rapid_feature_flags, + campaign_feature_flags=campaign_feature_flags, ) - - return job_model + return machine.run() def create_classification_job_definition( self, @@ -201,6 +163,7 @@ def create_classification_job_definition( confidence_threshold: float | None = None, quorum_threshold: int | None = None, settings: Sequence[RapidataSetting] | None = None, + failure_tolerance: float | None = None, private_metadata: list[dict[str, str]] | None = None, ) -> RapidataJobDefinition: """Create a classification job definition. @@ -228,6 +191,7 @@ def create_classification_job_definition( If provided, the classification datapoint will stop after the quorum is reached or at the number of responses, whatever happens first. Cannot be used together with confidence_threshold. settings (Sequence[RapidataSetting], optional): The list of settings for the classification. Defaults to []. Decides how the tasks should be shown. + failure_tolerance (float, optional): The fraction of datapoints allowed to fail while still creating the job definition (0.0-1.0). Defaults to None, which uses rapidata_config.upload.failureTolerance (default 0.0 = strict).\n 0.0 means any failed upload aborts creation so no incomplete definition is left behind; the failed datapoints can then be retried into the same dataset via the raised FailedUploadException.retry(). 1.0 creates the definition regardless of failures, as long as at least one datapoint uploads successfully. private_metadata (list[dict[str, str]], optional): Key-value string pairs for each datapoint. Defaults to None. If provided has to be the same length as datapoints.\n This will NOT be shown to the labelers but will be included in the result purely for your own reference. @@ -257,6 +221,7 @@ def create_classification_job_definition( confidence_threshold=confidence_threshold, quorum_threshold=quorum_threshold, settings=settings, + failure_tolerance=failure_tolerance, ) def create_compare_job_definition( @@ -272,6 +237,7 @@ def create_compare_job_definition( confidence_threshold: float | None = None, quorum_threshold: int | None = None, settings: Sequence[RapidataSetting] | None = None, + failure_tolerance: float | None = None, private_metadata: list[dict[str, str]] | None = None, ) -> RapidataJobDefinition: """Create a compare job definition. @@ -307,6 +273,7 @@ def create_compare_job_definition( If provided, the comparison datapoint will stop after the quorum is reached or at the number of responses, whatever happens first. Cannot be used together with confidence_threshold. settings (Sequence[RapidataSetting], optional): The list of settings for the comparison. Defaults to []. Decides how the tasks should be shown. + failure_tolerance (float, optional): The fraction of datapoints allowed to fail while still creating the job definition (0.0-1.0). Defaults to None, which uses rapidata_config.upload.failureTolerance (default 0.0 = strict).\n 0.0 means any failed upload aborts creation so no incomplete definition is left behind; the failed datapoints can then be retried into the same dataset via the raised FailedUploadException.retry(). 1.0 creates the definition regardless of failures, as long as at least one datapoint uploads successfully. private_metadata (list[dict[str, str]], optional): Key-value string pairs for each datapoint. Defaults to None.\n If provided has to be the same length as datapoints.\n This will NOT be shown to the labelers but will be included in the result purely for your own reference. @@ -343,6 +310,7 @@ def create_compare_job_definition( confidence_threshold=confidence_threshold, quorum_threshold=quorum_threshold, settings=settings, + failure_tolerance=failure_tolerance, ) def create_ranking_job_definition( @@ -357,6 +325,7 @@ def create_ranking_job_definition( contexts: list[str] | None = None, media_contexts: list[list[str]] | None = None, settings: Sequence[RapidataSetting] | None = None, + failure_tolerance: float | None = None, ) -> RapidataJobDefinition: """ Create a ranking job definition. @@ -381,6 +350,7 @@ def create_ranking_job_definition( Will be matched up with the datapoints using the list index. Use a single-element inner list for one image per ranking, or multiple entries to display several images. settings (Sequence[RapidataSetting], optional): The list of settings for the ranking. Defaults to []. Decides how the tasks should be shown. + failure_tolerance (float, optional): The fraction of datapoints allowed to fail while still creating the job definition (0.0-1.0). Defaults to None, which uses rapidata_config.upload.failureTolerance (default 0.0 = strict).\n 0.0 means any failed upload aborts creation so no incomplete definition is left behind; the failed datapoints can then be retried into the same dataset via the raised FailedUploadException.retry(). 1.0 creates the definition regardless of failures, as long as at least one datapoint uploads successfully. """ with tracer.start_as_current_span("JobManager.create_ranking_job"): if contexts and len(contexts) != len(datapoints): @@ -431,6 +401,7 @@ def create_ranking_job_definition( datapoints=datapoints_instances, responses_per_datapoint=responses_per_comparison, settings=settings, + failure_tolerance=failure_tolerance, ) def create_free_text_job_definition( @@ -443,6 +414,7 @@ def create_free_text_job_definition( contexts: list[str] | None = None, media_contexts: list[list[str]] | None = None, settings: Sequence[RapidataSetting] | None = None, + failure_tolerance: float | None = None, private_metadata: list[dict[str, str]] | None = None, ) -> RapidataJobDefinition: """Create a free text job definition. @@ -465,6 +437,7 @@ def create_free_text_job_definition( Will be matched up with the datapoints using the list index. Use a single-element inner list for one image per datapoint, or multiple entries to display several images. settings (Sequence[RapidataSetting], optional): The list of settings for the free text. Defaults to []. Decides how the tasks should be shown. + failure_tolerance (float, optional): The fraction of datapoints allowed to fail while still creating the job definition (0.0-1.0). Defaults to None, which uses rapidata_config.upload.failureTolerance (default 0.0 = strict).\n 0.0 means any failed upload aborts creation so no incomplete definition is left behind; the failed datapoints can then be retried into the same dataset via the raised FailedUploadException.retry(). 1.0 creates the definition regardless of failures, as long as at least one datapoint uploads successfully. private_metadata (list[dict[str, str]], optional): Key-value string pairs for each datapoint. Defaults to None.\n If provided has to be the same length as datapoints.\n This will NOT be shown to the labelers but will be included in the result purely for your own reference. @@ -485,6 +458,7 @@ def create_free_text_job_definition( datapoints=datapoints_instances, responses_per_datapoint=responses_per_datapoint, settings=settings, + failure_tolerance=failure_tolerance, ) def create_select_words_job_definition( @@ -495,6 +469,7 @@ def create_select_words_job_definition( sentences: list[str], responses_per_datapoint: int = 10, settings: Sequence[RapidataSetting] | None = None, + failure_tolerance: float | None = None, private_metadata: list[dict[str, str]] | None = None, ) -> RapidataJobDefinition: """Create a select words job definition. @@ -511,6 +486,7 @@ def create_select_words_job_definition( Must be the same length as datapoints. responses_per_datapoint (int, optional): The number of responses that will be collected per datapoint. Defaults to 10. settings (Sequence[RapidataSetting], optional): The list of settings for the select words. Defaults to []. Decides how the tasks should be shown. + failure_tolerance (float, optional): The fraction of datapoints allowed to fail while still creating the job definition (0.0-1.0). Defaults to None, which uses rapidata_config.upload.failureTolerance (default 0.0 = strict).\n 0.0 means any failed upload aborts creation so no incomplete definition is left behind; the failed datapoints can then be retried into the same dataset via the raised FailedUploadException.retry(). 1.0 creates the definition regardless of failures, as long as at least one datapoint uploads successfully. private_metadata (list[dict[str, str]], optional): Key-value string pairs for each datapoint. Defaults to None.\n If provided has to be the same length as datapoints.\n This will NOT be shown to the labelers but will be included in the result purely for your own reference. @@ -531,6 +507,7 @@ def create_select_words_job_definition( datapoints=datapoints_instances, responses_per_datapoint=responses_per_datapoint, settings=settings, + failure_tolerance=failure_tolerance, ) def create_locate_job_definition( @@ -542,6 +519,7 @@ def create_locate_job_definition( contexts: list[str] | None = None, media_contexts: list[list[str]] | None = None, settings: Sequence[RapidataSetting] | None = None, + failure_tolerance: float | None = None, private_metadata: list[dict[str, str]] | None = None, ) -> RapidataJobDefinition: """Create a locate job definition. @@ -561,6 +539,7 @@ def create_locate_job_definition( If provided has to be the same length as datapoints and will be shown in addition to the instruction. (Therefore will be different for each datapoint) Use a single-element inner list for one image per datapoint, or multiple entries to display several images. settings (Sequence[RapidataSetting], optional): The list of settings for the locate. Defaults to []. Decides how the tasks should be shown. + failure_tolerance (float, optional): The fraction of datapoints allowed to fail while still creating the job definition (0.0-1.0). Defaults to None, which uses rapidata_config.upload.failureTolerance (default 0.0 = strict).\n 0.0 means any failed upload aborts creation so no incomplete definition is left behind; the failed datapoints can then be retried into the same dataset via the raised FailedUploadException.retry(). 1.0 creates the definition regardless of failures, as long as at least one datapoint uploads successfully. private_metadata (list[dict[str, str]], optional): Key-value string pairs for each datapoint. Defaults to None.\n If provided has to be the same length as datapoints.\n This will NOT be shown to the labelers but will be included in the result purely for your own reference. @@ -580,6 +559,7 @@ def create_locate_job_definition( datapoints=datapoints_instances, responses_per_datapoint=responses_per_datapoint, settings=settings, + failure_tolerance=failure_tolerance, ) def create_draw_job_definition( @@ -591,6 +571,7 @@ def create_draw_job_definition( contexts: list[str] | None = None, media_contexts: list[list[str]] | None = None, settings: Sequence[RapidataSetting] | None = None, + failure_tolerance: float | None = None, private_metadata: list[dict[str, str]] | None = None, ) -> RapidataJobDefinition: """Create a draw job definition. @@ -610,6 +591,7 @@ def create_draw_job_definition( If provided has to be the same length as datapoints and will be shown in addition to the instruction. (Therefore will be different for each datapoint) Use a single-element inner list for one image per datapoint, or multiple entries to display several images. settings (Sequence[RapidataSetting], optional): The list of settings for the draw lines. Defaults to []. Decides how the tasks should be shown. + failure_tolerance (float, optional): The fraction of datapoints allowed to fail while still creating the job definition (0.0-1.0). Defaults to None, which uses rapidata_config.upload.failureTolerance (default 0.0 = strict).\n 0.0 means any failed upload aborts creation so no incomplete definition is left behind; the failed datapoints can then be retried into the same dataset via the raised FailedUploadException.retry(). 1.0 creates the definition regardless of failures, as long as at least one datapoint uploads successfully. private_metadata (list[dict[str, str]], optional): Key-value string pairs for each datapoint. Defaults to None.\n If provided has to be the same length as datapoints.\n This will NOT be shown to the labelers but will be included in the result purely for your own reference. @@ -629,6 +611,7 @@ def create_draw_job_definition( datapoints=datapoints_instances, responses_per_datapoint=responses_per_datapoint, settings=settings, + failure_tolerance=failure_tolerance, ) def _create_timestamp_job_definition( @@ -640,6 +623,7 @@ def _create_timestamp_job_definition( contexts: list[str] | None = None, media_contexts: list[list[str]] | None = None, settings: Sequence[RapidataSetting] | None = None, + failure_tolerance: float | None = None, private_metadata: list[dict[str, str]] | None = None, ) -> RapidataJobDefinition: """Create a timestamp job definition. @@ -662,6 +646,7 @@ def _create_timestamp_job_definition( If provided has to be the same length as datapoints and will be shown in addition to the instruction. (Therefore will be different for each datapoint) Use a single-element inner list for one image per datapoint, or multiple entries to display several images. settings (Sequence[RapidataSetting], optional): The list of settings for the timestamp. Defaults to []. Decides how the tasks should be shown. + failure_tolerance (float, optional): The fraction of datapoints allowed to fail while still creating the job definition (0.0-1.0). Defaults to None, which uses rapidata_config.upload.failureTolerance (default 0.0 = strict).\n 0.0 means any failed upload aborts creation so no incomplete definition is left behind; the failed datapoints can then be retried into the same dataset via the raised FailedUploadException.retry(). 1.0 creates the definition regardless of failures, as long as at least one datapoint uploads successfully. private_metadata (list[dict[str, str]], optional): Key-value string pairs for each datapoint. Defaults to None.\n If provided has to be the same length as datapoints.\n This will NOT be shown to the labelers but will be included in the result purely for your own reference. @@ -681,6 +666,7 @@ def _create_timestamp_job_definition( datapoints=datapoints_instances, responses_per_datapoint=responses_per_datapoint, settings=settings, + failure_tolerance=failure_tolerance, ) def get_job_definition_by_id(self, job_definition_id: str) -> RapidataJobDefinition: diff --git a/tests/rapidata_client/job/test_job_creation_state_machine.py b/tests/rapidata_client/job/test_job_creation_state_machine.py new file mode 100644 index 000000000..0057ab095 --- /dev/null +++ b/tests/rapidata_client/job/test_job_creation_state_machine.py @@ -0,0 +1,194 @@ +"""Tests for the job-definition creation state machine. + +The machine must persist the remote job definition only once the datapoint +upload lands within the configured failure tolerance, and a failed run must be +resumable into the *same* dataset (no second dataset, no second definition). +""" + +from __future__ import annotations + +from unittest.mock import MagicMock, patch + +import pytest + +from rapidata.rapidata_client.exceptions.failed_upload import FailedUpload +from rapidata.rapidata_client.exceptions.failed_upload_exception import ( + FailedUploadException, +) +from rapidata.rapidata_client.job._job_creation_state_machine import ( + JobDefinitionCreationMachine, +) + +MODULE = "rapidata.rapidata_client.job._job_creation_state_machine" +# _persist_definition imports this lazily; patch it so the mocked workflow/referee +# don't have to satisfy the pydantic input model. +JOB_INPUT = ( + "rapidata.api_client.models.create_job_definition_endpoint_input." + "CreateJobDefinitionEndpointInput" +) + + +def _make_openapi_service() -> MagicMock: + svc = MagicMock() + svc.environment = "rapidata.ai" + svc.dataset.dataset_api.dataset_post.return_value = MagicMock(dataset_id="ds-1") + svc.order.job_api.job_definition_post.return_value = MagicMock( + definition_id="def-1", pipeline_id="pipe-1" + ) + return svc + + +def _failed(item) -> FailedUpload: + return FailedUpload(item=item, error_message="boom", error_type="AssetUploadFailed") + + +def _run(svc, datapoints, tolerance, add_datapoints_side_effect): + """Build and run a machine with a mocked dataset whose add_datapoints is scripted.""" + dataset = MagicMock() + dataset.id = "ds-1" + dataset.add_datapoints.side_effect = add_datapoints_side_effect + + machine = JobDefinitionCreationMachine( + openapi_service=svc, + name="My Job", + workflow=MagicMock(), + datapoints=datapoints, + referee=MagicMock(), + failure_tolerance=tolerance, + ) + + with ( + patch(f"{MODULE}.RapidataDataset", return_value=dataset), + patch(f"{MODULE}.print_campaign_preview_qr_for_pipeline"), + patch(f"{MODULE}.print_job_definition_preview_link"), + patch(JOB_INPUT), + ): + return machine, dataset, machine.run() + + +def test_full_success_creates_definition(): + svc = _make_openapi_service() + dps = ["a", "b", "c"] + + machine, dataset, job_def = _run( + svc, dps, tolerance=0.0, add_datapoints_side_effect=[(dps, [])] + ) + + assert job_def.id == "def-1" + svc.dataset.dataset_api.dataset_post.assert_called_once() + svc.order.job_api.job_definition_post.assert_called_once() + + +def test_failure_within_tolerance_creates_definition(): + svc = _make_openapi_service() + dps = [str(i) for i in range(100)] + # 1/100 fails -> 1% == tolerance -> still creates the definition. + result = ([str(i) for i in range(99)], [_failed("99")]) + + _, _, job_def = _run(svc, dps, tolerance=0.01, add_datapoints_side_effect=[result]) + + assert job_def.id == "def-1" + svc.order.job_api.job_definition_post.assert_called_once() + + +def test_failure_exceeds_tolerance_raises_without_definition(): + svc = _make_openapi_service() + dps = [str(i) for i in range(100)] + # 5/100 fails, tolerance 0.0 -> no definition. + result = ([str(i) for i in range(95)], [_failed(str(i)) for i in range(95, 100)]) + + dataset = MagicMock() + dataset.id = "ds-1" + dataset.add_datapoints.side_effect = [result] + machine = JobDefinitionCreationMachine( + openapi_service=svc, + name="My Job", + workflow=MagicMock(), + datapoints=dps, + referee=MagicMock(), + failure_tolerance=0.0, + ) + + with ( + patch(f"{MODULE}.RapidataDataset", return_value=dataset), + patch(f"{MODULE}.print_campaign_preview_qr_for_pipeline"), + patch(f"{MODULE}.print_job_definition_preview_link"), + patch(JOB_INPUT), + ): + with pytest.raises(FailedUploadException) as excinfo: + machine.run() + + exc = excinfo.value + assert exc.machine is machine + assert exc.job_definition is None + svc.order.job_api.job_definition_post.assert_not_called() + + +def test_retry_reuses_dataset_and_persists_definition(): + svc = _make_openapi_service() + dps = ["a", "b", "c", "d"] + first = (["a", "b"], [_failed("c"), _failed("d")]) # 2/4 fail, tolerance 0.0 + second = (["c", "d"], []) # retry uploads the failed two successfully + + dataset = MagicMock() + dataset.id = "ds-1" + dataset.add_datapoints.side_effect = [first, second] + machine = JobDefinitionCreationMachine( + openapi_service=svc, + name="My Job", + workflow=MagicMock(), + datapoints=dps, + referee=MagicMock(), + failure_tolerance=0.0, + ) + + with ( + patch(f"{MODULE}.RapidataDataset", return_value=dataset) as dataset_cls, + patch(f"{MODULE}.print_campaign_preview_qr_for_pipeline"), + patch(f"{MODULE}.print_job_definition_preview_link"), + patch(JOB_INPUT), + ): + with pytest.raises(FailedUploadException) as excinfo: + machine.run() + + job_def = excinfo.value.retry() + + assert job_def.id == "def-1" + # Only one dataset ever created, and only one definition persisted. + dataset_cls.assert_called_once() + svc.dataset.dataset_api.dataset_post.assert_called_once() + svc.order.job_api.job_definition_post.assert_called_once() + # Retry re-uploaded ONLY the failed datapoints, into the same dataset. + assert dataset.add_datapoints.call_count == 2 + assert dataset.add_datapoints.call_args_list[1].args[0] == ["c", "d"] + + +def test_all_fail_never_creates_definition_even_at_full_tolerance(): + # Even failure_tolerance=1.0 must not create a definition over an empty + # dataset: a job with zero datapoints has nothing to label. + svc = _make_openapi_service() + dps = ["a", "b", "c"] + result = ([], [_failed("a"), _failed("b"), _failed("c")]) + + dataset = MagicMock() + dataset.id = "ds-1" + dataset.add_datapoints.side_effect = [result] + machine = JobDefinitionCreationMachine( + openapi_service=svc, + name="My Job", + workflow=MagicMock(), + datapoints=dps, + referee=MagicMock(), + failure_tolerance=1.0, + ) + + with ( + patch(f"{MODULE}.RapidataDataset", return_value=dataset), + patch(f"{MODULE}.print_campaign_preview_qr_for_pipeline"), + patch(f"{MODULE}.print_job_definition_preview_link"), + patch(JOB_INPUT), + ): + with pytest.raises(FailedUploadException): + machine.run() + + svc.order.job_api.job_definition_post.assert_not_called()