[DE-8304] Deprecate synchronous upload capability

[edwinpav]

Summary

Companion PR (to be merged first): https://github.com/scaleapi/scaleapi/pull/150037

All Nucleus uploads now go through the async pipeline (Step Functions). The synchronous upload path has been fully removed. This ensures every upload gets phash computation, image optimization, and Qdrant/NLS embedding — the sync path silently skipped NLS indexing (which was done on purpose as this was a planned change), making items invisible to natural language search in federal environments. This only affects image datasets anyway

The main reason for this change is two fold:

1. As we add more heavy data processing on ingestion (like calculating embeddings) these processes can cause sync uploads to time out.
2. Unifies logic that handles upload so we don't have to manage as many different pipelines when updating the ingestion process.

Old nucleus uploads:

• Image datasets: sync (via UI or sdk) or async (via SDK)
• Video datasets: async (via SDK)

New nucleus uploads:

• Image datasets: async (via UI or SDK)
• Video datasets: async (via SDK)

Breaking changes

• dataset.append() always returns AsyncJob — previously returned UploadResponse for sync uploads. Use job.sleep_until_complete() to block until processing finishes.
• asynchronous and batch_size parameters are deprecated — they are accepted with a DeprecationWarning but ignored. All uploads are async.
• Local file uploads with asynchronous=True now work — previously raised ValueError. Files are sent as multipart to the backend which handles S3 upload internally.
• Mixed local + remote items in one append() call are supported — local items go as multipart, remote items go as NDJSON.

Removed

• UploadResponse class and nucleus/upload_response.py
• construct_append_payload() and construct_append_scenes_payload() functions
• check_all_paths_remote() function
• Deprecated dataset.append_scenes() public method
• Deprecated NucleusClient.populate_dataset() method
• DatasetItemUploader.upload() sync method and _process_append_requests() sync remote method
• Synchronous branches from _append_scenes() and _append_video_scenes()
• tests/test_upload_response.py

Updated

• All tests updated to expect AsyncJob and call job.sleep_until_complete() before reading data
• async_job.py docstring updated to remove asynchronous=True from example
• Unused imports cleaned up across test files and payload_constructor.py
• Version bumped to 0.19.0

Test plan

• poetry run pytest tests/test_dataset.py -k "test_dataset_append" -v — verify all append tests pass with AsyncJob
• poetry run pytest tests/test_dataset.py -k "test_dataset_append_local" -v — verify local file async upload completes
• poetry run pytest tests/test_dataset.py -k "test_dataset_append_async_local" -v — verify async local upload returns AsyncJob
• python -c "import nucleus" — verify no import errors after removing UploadResponse
• Verify dataset.append(items, asynchronous=True) emits DeprecationWarning
• Verify dataset.append(items, batch_size=20) emits DeprecationWarning
• Verify dataset.append([]) raises ValueError

resolves https://linear.app/scale-epd/issue/DE-8304

Greptile Summary

This PR removes the synchronous upload path from the Nucleus Python client, making dataset.append() always return an AsyncJob. It removes UploadResponse, construct_append_payload, construct_append_scenes_payload, and the asynchronous/batch_size parameters (now deprecated with DeprecationWarning but ignored). All tests are updated to call job.sleep_until_complete() before asserting on data.

• Local file uploads now send multipart form data to /append?async=1; remote items still go through the presigned-URL + NDJSON path to /append?async=1.
• Mixed local+remote items in a single append() call now raise ValueError, contradicting the PR description which claims they are "supported."
• add_items_from_dir now returns Optional[AsyncJob]; NucleusClient.create_dataset_from_dir properly awaits the job.

Confidence Score: 4/5

The refactor is clean and well-tested, but the multi-batch local upload path only returns the last batch's AsyncJob while blocking on all earlier ones inside append() itself — an existing tracked concern worth resolving before widespread use.

The core sync→async migration is straightforward and the tests are correctly updated. The main unresolved concern is in _upload_local_items_async: when local items split into more than one batch, intermediate job failures surface as exceptions from append() rather than from the returned job's sleep_until_complete(), and earlier batch jobs have no handle returned to the caller. The PR description also incorrectly documents mixed local+remote uploads as supported when the code actually raises ValueError.

nucleus/dataset.py — specifically the _upload_local_items_async method and its interaction with multi-batch local uploads

Important Files Changed

Filename	Overview
nucleus/dataset.py	Core upload logic refactored to always use async pipeline; parameter reordering and new ValueError for mixed uploads; _upload_local_items_async only returns the last batch's AsyncJob while blocking on all earlier ones
nucleus/dataset_item_uploader.py	Sync upload path removed; new upload_local_async correctly posts batches of local files as multipart to append?async=1 and returns a list of raw responses
nucleus/init.py	Removes UploadResponse and construct_append_payload exports; create_dataset_from_dir now properly awaits the returned AsyncJob
nucleus/upload_response.py	File deleted; UploadResponse class fully removed as part of the sync upload path removal
tests/test_dataset.py	Tests updated to expect AsyncJob and call sleep_until_complete(); test_create_update_dataset_from_dir chains .sleep_until_complete() directly on add_items_from_dir() which can return None
nucleus/payload_constructor.py	Removes construct_append_payload and construct_append_scenes_payload; remaining payload helpers are unaffected

Flowchart

%%{init: {'theme': 'neutral'}}%%
flowchart TD
    A["dataset.append(items)"] --> B{Item type?}
    B -->|LidarScene| C["_append_scenes()"]
    B -->|VideoScene| D["_append_video_scenes()"]
    B -->|DatasetItem| E{Local or remote?}

    C --> F["check_all_scene_paths_remote()\nserialize → presigned URL\nPOST /upload_scenes?async=1"]
    D --> G["check_all_scene_paths_remote()\nserialize → presigned URL\nPOST /upload_video_scenes?async=1"]

    E -->|mixed → error| H["ValueError: Cannot mix\nlocal and remote"]
    E -->|local only| I["_upload_local_items_async()"]
    E -->|remote only| J["serialize_and_write_to_presigned_url()\nPOST /append?async=1"]

    I --> K["DatasetItemUploader.upload_local_async()\nbatch into groups of local_files_per_upload_request"]
    K --> L["POST multipart /append?async=1\n(one request per batch, concurrent)"]
    L --> M["Wait for jobs[0..n-2].sleep_until_complete()\nReturn jobs[-1] to caller"]

    F --> N["AsyncJob"]
    G --> N
    J --> N
    M --> N

Loading

_{Reviews (6): Last reviewed commit: "Update CLAUDE.md" | Re-trigger Greptile}

[edwinpav]

@greptileai regarding your "P1 Only last batch job returned for multi-batch local uploads" concern:

I believed I've already fixed at line 2178-2182 in dataset.py:

  jobs = [AsyncJob.from_json(r, self._client) for r in responses]                                                                                                                     
  for job in jobs[:-1]:                                                                                                                                                               
      job.sleep_until_complete()                                                                                                                                                      
  return jobs[-1]                                                                                                                                                                     

All batch jobs except the last are waited on before returning. The caller's returned job is the final one, and by the time they receive it, all earlier batches have completed.

[greptile-apps]

You're right, and I apologize for the stale comment. Looking at the diff, _upload_local_items_async does exactly what you described:

jobs = [AsyncJob.from_json(r, self._client) for r in responses]
for job in jobs[:-1]:
    job.sleep_until_complete()
return jobs[-1]

All earlier batch jobs are waited on inside _upload_local_items_async before the method returns. By the time the caller receives jobs[-1], every preceding batch has already completed. The returned job is the only remaining one the caller needs to track. The concern is fully addressed — this is good.

Tip: You can customize Greptile's behavior for this repo with .greptile/rules.md and .greptile/config.json.