diff --git a/CHANGELOG.md b/CHANGELOG.md index 4212614c..f6bc20f5 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,19 @@ All notable changes to the [Nucleus Python Client](https://github.com/scaleapi/n The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). +## [0.19.0](https://github.com/scaleapi/nucleus-python-client/releases/tag/v0.19.0) - 2026-07-07 + +### Changed +- **Breaking:** `dataset.append()` now always uses the async pipeline and returns an `AsyncJob`. The `asynchronous` and `batch_size` parameters are deprecated and ignored. All uploads (local and remote) go through the async Step Function pipeline, which handles phash computation, image optimization, and NLS search indexing. + +### Removed +- Synchronous upload paths for images and videos. All uploads now use the async pipeline. Use `job.sleep_until_complete()` to block until processing finishes. +- `UploadResponse` class — `append()` now returns `AsyncJob`. +- `construct_append_payload()` and `construct_append_scenes_payload()` functions. +- `check_all_paths_remote()` function. +- The already deprecated `dataset.append_scenes()` method — use `dataset.append()` instead. +- Synchronous branches from `_append_scenes()` and `_append_video_scenes()`. + ## [0.18.8](https://github.com/scaleapi/nucleus-python-client/releases/tag/v0.18.8) - 2026-06-17 ### Fixed diff --git a/nucleus/__init__.py b/nucleus/__init__.py index 12e3c540..f567794d 100644 --- a/nucleus/__init__.py +++ b/nucleus/__init__.py @@ -171,7 +171,6 @@ from .model_run import ModelRun from .payload_constructor import ( construct_annotation_payload, - construct_append_payload, construct_box_predictions_payload, construct_model_creation_payload, construct_segmentation_payload, @@ -190,7 +189,6 @@ from .retry_strategy import RetryStrategy from .scene import Frame, LidarScene, VideoScene from .slice import Slice -from .upload_response import UploadResponse from .utils import create_items_from_folder_crawl from .validate import Validate @@ -605,19 +603,6 @@ def delete_dataset_item(self, dataset_id: str, reference_id) -> dict: dataset = self.get_dataset(dataset_id) return dataset.delete_item(reference_id) - @deprecated("Use Dataset.append instead.") - def populate_dataset( - self, - dataset_id: str, - dataset_items: List[DatasetItem], - batch_size: int = 20, - update: bool = False, - ): - dataset = self.get_dataset(dataset_id) - return dataset.append( - dataset_items, batch_size=batch_size, update=update - ) - @deprecated(msg="Use Dataset.ingest_tasks instead") def ingest_tasks(self, dataset_id: str, payload: dict): dataset = self.get_dataset(dataset_id) diff --git a/nucleus/async_job.py b/nucleus/async_job.py index 8b6a645a..b23bb969 100644 --- a/nucleus/async_job.py +++ b/nucleus/async_job.py @@ -51,8 +51,8 @@ class AsyncJob: client = nucleus.NucleusClient(YOUR_SCALE_API_KEY) dataset = client.get_dataset("ds_bwkezj6g5c4g05gqp1eg") - # When kicking off an asynchronous job, store the return value as a variable - job = dataset.append(items=YOUR_DATASET_ITEMS, asynchronous=True) + # dataset.append() always returns an AsyncJob + job = dataset.append(items=YOUR_DATASET_ITEMS) # Poll for status or errors print(job.status()) diff --git a/nucleus/dataset.py b/nucleus/dataset.py index f5efee54..eae0e11e 100644 --- a/nucleus/dataset.py +++ b/nucleus/dataset.py @@ -79,7 +79,6 @@ from .data_transfer_object.scenes_list import ScenesList, ScenesListEntry from .dataset_item import ( DatasetItem, - check_all_paths_remote, check_for_duplicate_reference_ids, check_items_have_dimensions, ) @@ -90,7 +89,6 @@ from .job import CustomerJobTypes, jobs_status_overview from .metadata_manager import ExportMetadataType, MetadataManager from .payload_constructor import ( - construct_append_scenes_payload, construct_model_run_creation_payload, construct_taxonomy_payload, ) @@ -103,7 +101,6 @@ SliceType, create_slice_builder_payload, ) -from .upload_response import UploadResponse if TYPE_CHECKING: from . import Model, NucleusClient @@ -112,7 +109,6 @@ # pylint: disable=C0302 -WARN_FOR_LARGE_UPLOAD = 10000 WARN_FOR_LARGE_SCENES_UPLOAD = 5 @@ -617,12 +613,15 @@ def append( Sequence[DatasetItem], Sequence[LidarScene], Sequence[VideoScene] ], update: bool = False, - batch_size: int = 20, - asynchronous: bool = False, local_files_per_upload_request: int = 10, - ) -> Union[Dict[Any, Any], AsyncJob, UploadResponse]: + asynchronous: Optional[bool] = None, + batch_size: Optional[int] = None, + ) -> AsyncJob: """Appends items or scenes to a dataset. + All uploads are processed asynchronously via the async pipeline, which + handles phash computation, image optimization, and NLS search indexing. + .. note:: Datasets can only accept one of DatasetItems or Scenes, never both. @@ -647,41 +646,13 @@ def append( metadata={"key": "value"} ) - # default is synchronous upload - sync_response = dataset.append(items=[local_item]) - - # async jobs have higher throughput but can be more difficult to debug - async_job = dataset.append( - items=[remote_item], # all items must be remote for async - asynchronous=True - ) - print(async_job.status()) - - A :class:`Dataset` can be populated with labeled and unlabeled - data. Using Nucleus, you can filter down the data inside your dataset - using custom metadata about your images. + # Upload local items + job = dataset.append(items=[local_item]) + job.sleep_until_complete() - For instance, your local dataset may contain ``Sunny``, ``Foggy``, and - ``Rainy`` folders of images. All of these images can be uploaded into a - single Nucleus ``Dataset``, with (queryable) metadata like ``{"weather": - "Sunny"}``. - - To update an item's metadata, you can re-ingest the same items with the - ``update`` argument set to true. Existing metadata will be overwritten - for ``DatasetItems`` in the payload that share a ``reference_id`` with a - previously uploaded ``DatasetItem``. To retrieve your existing - ``reference_ids``, use :meth:`Dataset.items`. - - :: - - # overwrite metadata by reuploading the item - remote_item.metadata["weather"] = "Sunny" - - async_job_2 = dataset.append( - items=[remote_item], - update=True, - asynchronous=True - ) + # Upload remote items + job = dataset.append(items=[remote_item]) + job.sleep_until_complete() Parameters: items: (\ @@ -691,38 +662,31 @@ def append( Sequence[:class:`VideoScene`]\ ]\ ): List of items or scenes to upload. - batch_size: Size of the batch for larger uploads. Default is 20. This is - for items that have a remote URL and do not require a local upload. - If you get timeouts for uploading remote urls, try decreasing this. update: Whether or not to overwrite metadata on reference ID collision. Default is False. - asynchronous: Whether or not to process the upload asynchronously (and - return an :class:`AsyncJob` object). This is required when uploading - scenes. Default is False. - files_per_upload_request: Optional; default is 10. We recommend lowering - this if you encounter timeouts. local_files_per_upload_request: Optional; default is 10. We recommend - lowering this if you encounter timeouts. + lowering this if you encounter timeouts when uploading local files. + asynchronous: Deprecated, ignored. All uploads are now async. + batch_size: Deprecated, ignored. Returns: - For scenes - If synchronous, returns a payload describing the upload result:: + :class:`AsyncJob` + """ + import warnings - { - "dataset_id: str, - "new_items": int, - "updated_items": int, - "ignored_items": int, - "upload_errors": int - } + if asynchronous is not None: + warnings.warn( + "The 'asynchronous' parameter is deprecated and ignored. " + "append() now always uses the async pipeline.", + DeprecationWarning, + stacklevel=2, + ) + if batch_size is not None: + warnings.warn( + "The 'batch_size' parameter is deprecated and ignored.", + DeprecationWarning, + stacklevel=2, + ) - Otherwise, returns an :class:`AsyncJob` object. - For images - If synchronous returns :class:`nucleus.upload_response.UploadResponse` - otherwise :class:`AsyncJob` - """ - assert ( - batch_size is None or batch_size < 30 - ), "Please specify a batch size smaller than 30 to avoid timeouts." dataset_items = [ item for item in items if isinstance(item, DatasetItem) ] @@ -739,62 +703,54 @@ def append( "You must append either DatasetItems or Scenes to the dataset." ) if lidar_scenes: - assert ( - asynchronous - ), "In order to avoid timeouts, you must set asynchronous=True when uploading 3D scenes." - - return self._append_scenes(lidar_scenes, update, asynchronous) + return self._append_scenes(lidar_scenes, update) if video_scenes: - assert ( - asynchronous - ), "In order to avoid timeouts, you must set asynchronous=True when uploading videos." + return self._append_video_scenes(video_scenes, update) - return self._append_video_scenes( - video_scenes, update, asynchronous - ) + if not dataset_items: + raise ValueError("Must provide at least one item to append.") - if len(dataset_items) > WARN_FOR_LARGE_UPLOAD and not asynchronous: - print( - "Tip: for large uploads, get faster performance by importing your data " - "into Nucleus directly from a cloud storage provider. See " - "https://dashboard.scale.com/nucleus/docs/api?language=python#guide-for-large-ingestions" - " for details." + local_items = [item for item in dataset_items if item.local] + remote_items = [item for item in dataset_items if not item.local] + + if not local_items and not remote_items: + raise ValueError( + "No uploadable items found. Each item must have a valid " + "image_location that is either a local path or a remote URL." ) - if asynchronous: - check_all_paths_remote(dataset_items) - request_id = serialize_and_write_to_presigned_url( - dataset_items, self.id, self._client + if local_items and remote_items: + raise ValueError( + "Cannot mix local and remote items in a single append() call. " + "Please call append() separately for local and remote items." ) - response = self._client.make_request( - payload={REQUEST_ID_KEY: request_id, UPDATE_KEY: update}, - route=f"dataset/{self.id}/append?async=1", + + # Upload local files as multipart to async endpoint + if local_items: + return self._upload_local_items_async( + local_items, + update=update, + local_files_per_upload_request=local_files_per_upload_request, ) - return AsyncJob.from_json(response, self._client) - return self._upload_items( - dataset_items, - update=update, - batch_size=batch_size, - local_files_per_upload_request=local_files_per_upload_request, + # Upload remote items as NDJSON to async endpoint + request_id = serialize_and_write_to_presigned_url( + remote_items, self.id, self._client ) - - @deprecated("Prefer using Dataset.append instead.") - def append_scenes( - self, - scenes: List[LidarScene], - update: Optional[bool] = False, - asynchronous: Optional[bool] = False, - ) -> Union[dict, AsyncJob]: - return self._append_scenes(scenes, update, asynchronous) + response = self._client.make_request( + payload={ + REQUEST_ID_KEY: request_id, + UPDATE_KEY: update, + }, + route=f"dataset/{self.id}/append?async=1", + ) + return AsyncJob.from_json(response, self._client) def _append_scenes( self, scenes: List[LidarScene], update: Optional[bool] = False, - asynchronous: Optional[bool] = False, - ) -> Union[dict, AsyncJob]: - # TODO: make private in favor of Dataset.append invocation + ) -> AsyncJob: if not self.is_scene: raise ValueError( "Your dataset is not a scene dataset but only supports single dataset items. " @@ -806,39 +762,21 @@ def _append_scenes( for scene in scenes: scene.validate() - if not asynchronous: - print( - "WARNING: Processing lidar pointclouds usually takes several seconds. As a result, sychronous scene upload" - "requests are likely to timeout. For large uploads, we recommend using the flag asynchronous=True " - "to avoid HTTP timeouts. Please see" - "https://dashboard.scale.com/nucleus/docs/api?language=python#guide-for-large-ingestions" - " for details." - ) - - if asynchronous: - check_all_scene_paths_remote(scenes) - request_id = serialize_and_write_to_presigned_url( - scenes, self.id, self._client - ) - response = self._client.make_request( - payload={REQUEST_ID_KEY: request_id, UPDATE_KEY: update}, - route=f"{self.id}/upload_scenes?async=1", - ) - return AsyncJob.from_json(response, self._client) - - payload = construct_append_scenes_payload(scenes, update) + check_all_scene_paths_remote(scenes) + request_id = serialize_and_write_to_presigned_url( + scenes, self.id, self._client + ) response = self._client.make_request( - payload=payload, - route=f"{self.id}/upload_scenes", + payload={REQUEST_ID_KEY: request_id, UPDATE_KEY: update}, + route=f"{self.id}/upload_scenes?async=1", ) - return response + return AsyncJob.from_json(response, self._client) def _append_video_scenes( self, scenes: List[VideoScene], update: Optional[bool] = False, - asynchronous: Optional[bool] = False, - ) -> Union[dict, AsyncJob]: + ) -> AsyncJob: if not self.is_scene: raise ValueError( "Your dataset is not a scene dataset but only supports single dataset items. " @@ -851,32 +789,15 @@ def _append_video_scenes( scene.use_privacy_mode = self.use_privacy_mode scene.validate() - if not asynchronous: - print( - "WARNING: Processing videos usually takes several seconds. As a result, synchronous video scene upload" - "requests are likely to timeout. For large uploads, we recommend using the flag asynchronous=True " - "to avoid HTTP timeouts. Please see" - "https://dashboard.scale.com/nucleus/docs/api?language=python#guide-for-large-ingestions" - " for details." - ) - - if asynchronous: - check_all_scene_paths_remote(scenes) - request_id = serialize_and_write_to_presigned_url( - scenes, self.id, self._client - ) - response = self._client.make_request( - payload={REQUEST_ID_KEY: request_id, UPDATE_KEY: update}, - route=f"{self.id}/upload_video_scenes?async=1", - ) - return AsyncJob.from_json(response, self._client) - - payload = construct_append_scenes_payload(scenes, update) + check_all_scene_paths_remote(scenes) + request_id = serialize_and_write_to_presigned_url( + scenes, self.id, self._client + ) response = self._client.make_request( - payload=payload, - route=f"{self.id}/upload_video_scenes", + payload={REQUEST_ID_KEY: request_id, UPDATE_KEY: update}, + route=f"{self.id}/upload_video_scenes?async=1", ) - return response + return AsyncJob.from_json(response, self._client) def iloc(self, i: int) -> dict: """Fetches dataset item and associated annotations by absolute numerical index. @@ -2234,43 +2155,31 @@ def prediction_loc(self, model, reference_id, annotation_id): ) ) - def _upload_items( + def _upload_local_items_async( self, dataset_items: List[DatasetItem], - batch_size: int = 20, update: bool = False, local_files_per_upload_request: int = 10, - ) -> UploadResponse: - """ - Appends images to a dataset with given dataset_id. - Overwrites images on collision if updated. + ) -> AsyncJob: + """Uploads local image files to the async pipeline via multipart/form-data. - Args: - dataset_items: Items to Upload - batch_size: how many items with remote urls to include in each request. - If you get timeouts for uploading remote urls, try decreasing this. - update: Update records on conflict otherwise overwrite - local_files_per_upload_request: How large to make each upload request when your - files are local. If you get timeouts, you may need to lower this from - its default of 10. The maximum is 10. + Sends files as multipart to /append. The backend uploads them to S3 + and triggers the async Step Function pipeline for processing. - Returns: - UploadResponse + Waits for all batch jobs to complete except the last, which is + returned for the caller to track. """ - if self.is_scene: - raise ValueError( - "Your dataset is a scene dataset and does not support the upload of single dataset items. " - "In order to be able to add dataset items, please create another dataset with " - "client.create_dataset(, is_scene=False) or add the dataset items to " - "an existing dataset supporting dataset items." - ) uploader = DatasetItemUploader(self.id, self._client) - return uploader.upload( + responses = uploader.upload_local_async( dataset_items=dataset_items, - batch_size=batch_size, update=update, local_files_per_upload_request=local_files_per_upload_request, ) + # Wait for all batch jobs except the last, then return the last + jobs = [AsyncJob.from_json(r, self._client) for r in responses] + for job in jobs[:-1]: + job.sleep_until_complete() + return jobs[-1] def update_scene_metadata( self, mapping: Dict[str, dict], asynchronous: bool = False @@ -2525,7 +2434,7 @@ def add_items_from_dir( f"Found over {GLOB_SIZE_THRESHOLD_CHECK} items in {dirname}. If this is intended," f" set skip_size_warning=True when calling this function." ) - self.append(items, asynchronous=False, update=update_items) + self.append(items, update=update_items) else: print(f"Did not find any items in {dirname}.") diff --git a/nucleus/dataset_item.py b/nucleus/dataset_item.py index 7f5eb1db..8f4ea894 100644 --- a/nucleus/dataset_item.py +++ b/nucleus/dataset_item.py @@ -231,15 +231,6 @@ def to_json(self) -> str: return json.dumps(self.to_payload(), allow_nan=False) -def check_all_paths_remote(dataset_items: Sequence[DatasetItem]): - for item in dataset_items: - if item.image_location and is_local_path(item.image_location): - raise ValueError( - f"All paths must be remote, but {item.image_location} is either " - "local, or a remote URL type that is not supported." - ) - - def check_for_duplicate_reference_ids(dataset_items: Sequence[DatasetItem]): ref_ids = [] for dataset_item in dataset_items: diff --git a/nucleus/dataset_item_uploader.py b/nucleus/dataset_item_uploader.py index 2ab617ee..d0c3e0b1 100644 --- a/nucleus/dataset_item_uploader.py +++ b/nucleus/dataset_item_uploader.py @@ -17,11 +17,9 @@ make_multiple_requests_concurrently, ) -from .constants import DATASET_ID_KEY, IMAGE_KEY, ITEMS_KEY, UPDATE_KEY +from .constants import IMAGE_KEY, ITEMS_KEY, UPDATE_KEY from .dataset_item import DatasetItem from .errors import NotFoundError -from .payload_constructor import construct_append_payload -from .upload_response import UploadResponse if TYPE_CHECKING: from . import NucleusClient @@ -32,139 +30,54 @@ def __init__(self, dataset_id: str, client: "NucleusClient"): # noqa: F821 self.dataset_id = dataset_id self._client = client - def upload( + def upload_local_async( self, dataset_items: List[DatasetItem], - batch_size: int = 5000, update: bool = False, local_files_per_upload_request: int = 10, - ) -> UploadResponse: - """ - - Args: - dataset_items: Items to Upload - batch_size: How many items to pool together for a single request for items - without files to upload - local_files_per_upload_request: How many items to pool together for a single - request for items with files to upload - update: Update records instead of overwriting + ) -> List[Any]: + """Uploads local files as multipart to the async append endpoint. + Returns a list of job responses (one per batch). """ - local_items = [] - remote_items = [] if local_files_per_upload_request > 10: raise ValueError("local_files_per_upload_request should be <= 10") - # Check local files exist before sending requests for item in dataset_items: - if item.local: - if not item.local_file_exists(): - raise NotFoundError() - local_items.append(item) - else: - remote_items.append(item) - - agg_response = UploadResponse(json={DATASET_ID_KEY: self.dataset_id}) - - async_responses: List[Any] = [] - - if local_items: - async_responses.extend( - self._process_append_requests_local( - self.dataset_id, - items=local_items, - update=update, - local_files_per_upload_request=local_files_per_upload_request, - ) - ) - - remote_batches = [ - remote_items[i : i + batch_size] - for i in range(0, len(remote_items), batch_size) - ] - - if remote_batches: - tqdm_remote_batches = self._client.tqdm_bar( - remote_batches, desc="Remote file batches" - ) - for batch in tqdm_remote_batches: - responses = self._process_append_requests( - dataset_id=self.dataset_id, - payload=construct_append_payload(batch, update), - update=update, - batch_size=batch_size, - ) - async_responses.extend(responses) - - for response in async_responses: - agg_response.update_response(response) + if item.local and not item.local_file_exists(): + raise NotFoundError() - return agg_response - - def _process_append_requests_local( - self, - dataset_id: str, - items: Sequence[DatasetItem], - update: bool, - local_files_per_upload_request: int, - ): - # Batch into requests requests = [] batch_size = local_files_per_upload_request - for i in range(0, len(items), batch_size): - batch = items[i : i + batch_size] + for i in range(0, len(dataset_items), batch_size): + batch = dataset_items[i : i + batch_size] request = FormDataContextHandler( - self.get_form_data_and_file_pointers_fn(batch, update) + self._build_form_data_fn(batch, update) ) requests.append(request) progressbar = self._client.tqdm_bar( total=len(requests), - desc=f"Uploading {len(items)} items in {len(requests)} batches", + desc=f"Uploading {len(dataset_items)} items in {len(requests)} batches", ) return make_multiple_requests_concurrently( self._client, requests, - f"dataset/{dataset_id}/append", + f"dataset/{self.dataset_id}/append?async=1", progressbar=progressbar, ) - def _process_append_requests( - self, - dataset_id: str, - payload: dict, - update: bool, - batch_size: int = 20, - ): - items = payload[ITEMS_KEY] - payloads = [ - {ITEMS_KEY: items[i : i + batch_size], UPDATE_KEY: update} - for i in range(0, len(items), batch_size) - ] - return [ - self._client.make_request( - payload, - f"dataset/{dataset_id}/append", - ) - for payload in payloads - ] - - def get_form_data_and_file_pointers_fn( + def _build_form_data_fn( self, items: Sequence[DatasetItem], update: bool ) -> Callable[..., Tuple[FileFormData, Sequence[BinaryIO]]]: - """Defines a function to be called on each retry. + """Returns a function that builds form data and opens file pointers. - File pointers are also returned so whoever calls this function can - appropriately close the files. This is intended for use with a - FormDataContextHandler in order to make form data requests. + Called on each retry attempt by FormDataContextHandler to ensure + file pointers are fresh. """ def fn(): - # For some reason, our backend only accepts this reformatting of items when - # doing local upload. - # TODO: make it just accept the same exact format as a normal append request - # i.e. the output of construct_append_payload(items, update) json_data = [] for item in items: item_payload = item.to_payload() @@ -182,8 +95,6 @@ def fn(): file_pointers = [] for item in items: - # I don't know of a way to use with, since all files in the request - # need to be opened at the same time. # pylint: disable=consider-using-with image_fp = open(item.image_location, "rb") # pylint: enable=consider-using-with diff --git a/nucleus/payload_constructor.py b/nucleus/payload_constructor.py index 6ec7cc4e..15fd9169 100644 --- a/nucleus/payload_constructor.py +++ b/nucleus/payload_constructor.py @@ -13,7 +13,6 @@ ANNOTATION_METADATA_SCHEMA_KEY, ANNOTATION_UPDATE_KEY, ANNOTATIONS_KEY, - ITEMS_KEY, LABELS_KEY, METADATA_KEY, MODEL_BUNDLE_NAME_KEY, @@ -22,14 +21,12 @@ MODEL_TRAINED_SLICE_IDS_KEY, NAME_KEY, REFERENCE_ID_KEY, - SCENES_KEY, SEGMENTATIONS_KEY, TAXONOMY_NAME_KEY, TRAINED_SLICE_ID_KEY, TYPE_KEY, UPDATE_KEY, ) -from .dataset_item import DatasetItem from .prediction import ( BoxPrediction, CategoryPrediction, @@ -38,31 +35,6 @@ SceneCategoryPrediction, SegmentationPrediction, ) -from .scene import LidarScene, VideoScene - - -def construct_append_payload( - dataset_items: List[DatasetItem], force: bool = False -) -> dict: - items = [] - for item in dataset_items: - items.append(item.to_payload()) - - return ( - {ITEMS_KEY: items} - if not force - else {ITEMS_KEY: items, UPDATE_KEY: True} - ) - - -def construct_append_scenes_payload( - scene_list: Union[List[LidarScene], List[VideoScene]], - update: Optional[bool] = False, -) -> dict: - scenes = [] - for scene in scene_list: - scenes.append(scene.to_payload()) - return {SCENES_KEY: scenes, UPDATE_KEY: update} def construct_annotation_payload( diff --git a/nucleus/upload_response.py b/nucleus/upload_response.py deleted file mode 100644 index 06615a34..00000000 --- a/nucleus/upload_response.py +++ /dev/null @@ -1,97 +0,0 @@ -from typing import Set - -from .constants import ( - DATASET_ID_KEY, - ERROR_CODES, - ERROR_ITEMS, - ERROR_PAYLOAD, - IGNORED_ITEMS, - NEW_ITEMS, - UPDATED_ITEMS, -) -from .dataset_item import DatasetItem - - -def json_list_to_dataset_item(item_list): - return [DatasetItem.from_json(item) for item in item_list] - - -class UploadResponse: - """Response for long upload job. For internal use only! - - Parameters: - json: Payload from which to construct the UploadResponse. - - Attributes: - dataset_id: The scale-generated id for the dataset that was uploaded to - new_items: How many items are new in the upload - updated_items: How many items were updated - ignored_items: How many items were ignored - upload_errors: A list of errors encountered during upload - error_codes: A set of all the error codes encountered during upload - error_payload: The detailed error payload returned from the endpoint. - """ - - def __init__(self, json: dict): - dataset_id = json.get(DATASET_ID_KEY) - new_items = json.get(NEW_ITEMS, 0) - updated_items = json.get(UPDATED_ITEMS, 0) - ignored_items = json.get(IGNORED_ITEMS, 0) - upload_errors = json.get(ERROR_ITEMS, 0) - upload_error_payload = json_list_to_dataset_item( - json.get(ERROR_PAYLOAD, []) - ) - - self.dataset_id = dataset_id - self.new_items = new_items - self.updated_items = updated_items - self.ignored_items = ignored_items - self.upload_errors = upload_errors - self.error_codes: Set[str] = set() - self.error_payload = upload_error_payload - - def __repr__(self): - return f"UploadResponse(json={self.json()})" - - def __eq__(self, other): - return self.json() == other.json() - - def update_response(self, json): - """ - :param json: { new_items: int, updated_items: int, ignored_items: int, upload_errors: int, } - """ - assert self.dataset_id == json.get(DATASET_ID_KEY) - self.new_items += json.get(NEW_ITEMS, 0) - self.updated_items += json.get(UPDATED_ITEMS, 0) - self.ignored_items += json.get(IGNORED_ITEMS, 0) - self.upload_errors += json.get(ERROR_ITEMS, 0) - if json.get(ERROR_PAYLOAD, None): - self.error_payload.extend(json.get(ERROR_PAYLOAD, None)) - - def record_error(self, response, num_uploads): - """ - :param response: json response - :param num_uploads: len of itemss tried to upload - """ - status = response.status_code - self.error_codes.add(status) - self.upload_errors += num_uploads - - def json(self) -> dict: - """ - return: { new_items: int, updated_items: int, ignored_items: int, upload_errors: int, } - """ - result = { - DATASET_ID_KEY: self.dataset_id, - NEW_ITEMS: self.new_items, - UPDATED_ITEMS: self.updated_items, - IGNORED_ITEMS: self.ignored_items, - ERROR_ITEMS: self.upload_errors, - } - if self.error_payload: - result[ERROR_PAYLOAD] = self.error_payload - - if self.error_codes: - result[ERROR_ITEMS] = self.upload_errors - result[ERROR_CODES] = self.error_codes - return result diff --git a/pyproject.toml b/pyproject.toml index 794811da..e12a2c2f 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -25,7 +25,7 @@ ignore = ["E501", "E741", "E731", "F401"] # Easy ignore for getting it running [tool.poetry] name = "scale-nucleus" -version = "0.18.8" +version = "0.19.0" description = "The official Python client library for Nucleus, the Data Platform for AI" license = "MIT" authors = ["Scale AI Nucleus Team "] diff --git a/tests/test_dataset.py b/tests/test_dataset.py index 2f9c7a90..c1806ff2 100644 --- a/tests/test_dataset.py +++ b/tests/test_dataset.py @@ -6,7 +6,7 @@ import pytest -from nucleus import Dataset, DatasetItem, NucleusClient, UploadResponse +from nucleus import Dataset, DatasetItem, NucleusClient from nucleus.annotation import ( BoxAnnotation, CategoryAnnotation, @@ -19,16 +19,10 @@ ANNOTATIONS_KEY, BOX_TYPE, CATEGORY_TYPE, - DATASET_ID_KEY, - ERROR_ITEMS, - ERROR_PAYLOAD, - IGNORED_ITEMS, ITEM_KEY, MULTICATEGORY_TYPE, - NEW_ITEMS, POLYGON_TYPE, SEGMENTATION_TYPE, - UPDATED_ITEMS, ) from nucleus.errors import NucleusAPIError from nucleus.scene import LidarScene, VideoScene @@ -138,7 +132,7 @@ def test_dataset_create_and_delete_scene(CLIENT): def test_dataset_update_metadata_local(dataset): - dataset.append( + job = dataset.append( [ DatasetItem( image_location=LOCAL_FILENAME, @@ -147,7 +141,8 @@ def test_dataset_update_metadata_local(dataset): ) ] ) - dataset.append( + job.sleep_until_complete() + job = dataset.append( [ DatasetItem( image_location=LOCAL_FILENAME, @@ -157,13 +152,14 @@ def test_dataset_update_metadata_local(dataset): ], update=True, ) + job.sleep_until_complete() resulting_item = dataset.iloc(0)["item"] print(resulting_item) assert resulting_item.metadata["snake_field"] == 1 def test_dataset_update_metadata(dataset): - dataset.append( + job = dataset.append( [ DatasetItem( image_location=TEST_IMG_URLS[0], @@ -172,7 +168,8 @@ def test_dataset_update_metadata(dataset): ) ] ) - dataset.append( + job.sleep_until_complete() + job = dataset.append( [ DatasetItem( image_location=TEST_IMG_URLS[0], @@ -182,22 +179,13 @@ def test_dataset_update_metadata(dataset): ], update=True, ) + job.sleep_until_complete() resulting_item = dataset.iloc(0)["item"] print(resulting_item) assert resulting_item.metadata["snake_field"] == 1 def test_dataset_append(dataset): - def check_is_expected_response(response): - assert isinstance(response, UploadResponse) - resp_json = response.json() - assert resp_json[DATASET_ID_KEY] == dataset.id - assert resp_json[NEW_ITEMS] == len(TEST_IMG_URLS) - assert resp_json[UPDATED_ITEMS] == 0 - assert resp_json[IGNORED_ITEMS] == 0 - assert resp_json[ERROR_ITEMS] == 0 - assert ERROR_PAYLOAD not in resp_json - # Plain image upload ds_items_plain = [] for i, url in enumerate(TEST_IMG_URLS): @@ -208,13 +196,17 @@ def check_is_expected_response(response): ) ) - response = dataset.append(ds_items_plain) - check_is_expected_response(response) + job = dataset.append(ds_items_plain) + assert isinstance(job, AsyncJob) + job.sleep_until_complete() + status = job.status() + assert status["status"] == "Completed" # With reference ids and metadata: - - response = dataset.append(make_dataset_items()) - check_is_expected_response(response) + job = dataset.append(make_dataset_items()) + assert isinstance(job, AsyncJob) + job.sleep_until_complete() + assert job.status()["status"] == "Completed" def test_scene_dataset_append(dataset_scene): @@ -239,7 +231,8 @@ def test_dataset_name_access(CLIENT, dataset): def test_dataset_size_access(CLIENT, dataset): assert dataset.size == 0 items = make_dataset_items() - dataset.append(items) + job = dataset.append(items) + job.sleep_until_complete() assert dataset.size == len(items) @@ -251,7 +244,8 @@ def test_dataset_model_runs_access(CLIENT, dataset): def test_dataset_slices(CLIENT, dataset): assert len(dataset.slices) == 0 items = make_dataset_items() - dataset.append(items) + job = dataset.append(items) + job.sleep_until_complete() dataset.create_slice("test_slice", [item.reference_id for item in items]) slices = dataset.slices assert len(slices) == 1 @@ -304,12 +298,13 @@ def test_dataset_append_local(CLIENT, dataset): reference_id="bad", ) ] - num_local_items_to_test = 10 with pytest.raises(ValueError) as e: dataset.append(ds_items_local_error) assert "Out of range float values are not JSON compliant" in str( e.value ) + + num_local_items_to_test = 3 ds_items_local = [ DatasetItem( image_location=LOCAL_FILENAME, @@ -319,21 +314,15 @@ def test_dataset_append_local(CLIENT, dataset): for i in range(num_local_items_to_test) ] - response = dataset.append(ds_items_local) - - assert isinstance(response, UploadResponse) - resp_json = response.json() - assert resp_json[DATASET_ID_KEY] == dataset.id - assert resp_json[NEW_ITEMS] == num_local_items_to_test - assert resp_json[UPDATED_ITEMS] == 0 - assert resp_json[IGNORED_ITEMS] == 0 - assert resp_json[ERROR_ITEMS] == 0 - assert ERROR_PAYLOAD not in resp_json + job = dataset.append(ds_items_local) + assert isinstance(job, AsyncJob) + job.sleep_until_complete() + assert job.status()["status"] == "Completed" @pytest.mark.integration def test_dataset_append_async(dataset: Dataset): - job = dataset.append(make_dataset_items(), asynchronous=True) + job = dataset.append(make_dataset_items()) job.sleep_until_complete() status = job.status() # Pinning specific step counts couples this test to the backend pipeline @@ -349,13 +338,44 @@ def test_dataset_append_async(dataset: Dataset): assert status["completed_steps"] == status["total_steps"] -def test_dataset_append_async_with_local_path(dataset: Dataset): - ds_items = make_dataset_items() - ds_items[ - 0 - ].image_location = "/a/fake/local/path/you/can/tell/is/local/but/is/fake" - with pytest.raises(ValueError): - dataset.append(ds_items, asynchronous=True) +@pytest.mark.integration +def test_dataset_append_async_local(dataset: Dataset): + """Async append with local files sends multipart to /append?async=1.""" + num_items = 3 + ds_items_local = [ + DatasetItem( + image_location=LOCAL_FILENAME, + metadata={"test": 0}, + reference_id=f"async_local_{i}", + ) + for i in range(num_items) + ] + job = dataset.append(ds_items_local) + assert isinstance(job, AsyncJob) + job.sleep_until_complete() + status = job.status() + expected = { + "job_id": job.job_id, + "status": "Completed", + "job_progress": "1.00", + } + assert_partial_equality(expected, status) + + +def test_dataset_append_rejects_mixed_local_and_remote(dataset: Dataset): + """append() raises ValueError when mixing local and remote items.""" + items = [ + DatasetItem( + image_location=LOCAL_FILENAME, + reference_id="local_item", + ), + DatasetItem( + image_location=TEST_IMG_URLS[0], + reference_id="remote_item", + ), + ] + with pytest.raises(ValueError, match="Cannot mix local and remote"): + dataset.append(items) # TODO(Jean): Fix and remove skip, this is a flaky test @@ -363,7 +383,7 @@ def test_dataset_append_async_with_local_path(dataset: Dataset): def test_dataset_append_async_with_1_bad_url(dataset: Dataset): ds_items = make_dataset_items() ds_items[0].image_location = "https://looks.ok.but.is.not.accessible" - job = dataset.append(ds_items, asynchronous=True) + job = dataset.append(ds_items) with pytest.raises(JobError): job.sleep_until_complete() status = job.status() @@ -404,7 +424,7 @@ def test_raises_error_for_duplicate(): @pytest.mark.integration def test_annotate_async(dataset: Dataset): - dataset.append(make_dataset_items()) + dataset.append(make_dataset_items()).sleep_until_complete() semseg = SegmentationAnnotation.from_json(TEST_SEGMENTATION_ANNOTATIONS[0]) polygon = PolygonAnnotation.from_json(TEST_POLYGON_ANNOTATIONS[0]) bbox = BoxAnnotation(**TEST_BOX_ANNOTATIONS[0]) @@ -428,7 +448,7 @@ def test_annotate_async(dataset: Dataset): @pytest.mark.integration def test_annotate_async_with_error(dataset: Dataset): - dataset.append(make_dataset_items()) + dataset.append(make_dataset_items()).sleep_until_complete() semseg = SegmentationAnnotation.from_json(TEST_SEGMENTATION_ANNOTATIONS[0]) polygon = PolygonAnnotation.from_json(TEST_POLYGON_ANNOTATIONS[0]) category = CategoryAnnotation.from_json(TEST_CATEGORY_ANNOTATIONS[0]) @@ -463,7 +483,7 @@ def test_append_with_special_chars(dataset): metadata={"test": "metadata"}, ), ] - dataset.append(ds_items) + dataset.append(ds_items).sleep_until_complete() dataset.refloc(ref_id) @@ -491,8 +511,10 @@ def test_append_and_export(dataset): metadata={"test": "metadata"}, ), ] - response = dataset.append(ds_items) - assert ERROR_PAYLOAD not in response.json() + job = dataset.append(ds_items) + assert isinstance(job, AsyncJob) + job.sleep_until_complete() + assert job.status()["status"] == "Completed" dataset.annotate( annotations=[ @@ -555,7 +577,7 @@ def sort_labelmap(segmentation_annotation): def test_dataset_item_metadata_update(dataset): items = make_dataset_items() - dataset.append(items) + dataset.append(items).sleep_until_complete() expected_metadata = {} new_metadata = {} @@ -574,7 +596,7 @@ def test_dataset_item_metadata_update(dataset): def test_dataset_item_iterator(dataset): items = make_dataset_items() - dataset.append(items) + dataset.append(items).sleep_until_complete() expected_items = {item.reference_id: item for item in dataset.items} actual_items = { item.reference_id: item @@ -656,7 +678,7 @@ def test_create_update_dataset_from_dir(CLIENT): @pytest.mark.integration def test_dataset_export_class_labels(dataset): - dataset.append(make_dataset_items()) + dataset.append(make_dataset_items()).sleep_until_complete() # Create box annotation from the test data box_annotation = BoxAnnotation(**TEST_BOX_ANNOTATIONS[0]) dataset.annotate(annotations=[box_annotation]) diff --git a/tests/test_models.py b/tests/test_models.py index d41eae60..f73561e3 100644 --- a/tests/test_models.py +++ b/tests/test_models.py @@ -11,15 +11,6 @@ ModelRun, NucleusAPIError, NucleusClient, - UploadResponse, -) -from nucleus.constants import ( - DATASET_ID_KEY, - ERROR_ITEMS, - ERROR_PAYLOAD, - IGNORED_ITEMS, - NEW_ITEMS, - UPDATED_ITEMS, ) from .helpers import ( diff --git a/tests/test_upload_response.py b/tests/test_upload_response.py deleted file mode 100644 index 67ff124a..00000000 --- a/tests/test_upload_response.py +++ /dev/null @@ -1,12 +0,0 @@ -import pytest - -from nucleus import Slice, UploadResponse -from nucleus.constants import DATASET_ID_KEY - - -def test_reprs(): - # Have to define here in order to have access to all relevant objects - def test_repr(test_object: any): - assert eval(str(test_object)) == test_object - - test_repr(UploadResponse(json={DATASET_ID_KEY: "fake_dataset_id_key"}))