[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, routing all dataset.append() calls through the async Step Function pipeline. The main motivation is ensuring every upload gets phash computation, image optimisation, and NLS search indexing — capabilities that the sync path silently skipped.

• append() now always returns AsyncJob; the deprecated asynchronous and batch_size parameters emit DeprecationWarning and are ignored. Local file uploads go as multipart to /append?async=1; remote items are serialised to a presigned URL and submitted the existing way.
• Mixed local+remote in a single call now raises ValueError (the PR description claims they are "supported", but the code and the new test_dataset_append_rejects_mixed_local_and_remote test both confirm they are rejected).
• UploadResponse, construct_append_payload, check_all_paths_remote, and the deprecated populate_dataset / append_scenes wrappers are all removed; version bumped to 0.19.0.

Confidence Score: 4/5

Safe to merge after addressing multi-batch local upload tracking: callers uploading 11+ local files only receive the last batch's AsyncJob, leaving earlier batches untracked.

The sync-to-async migration is clean and well-tested. The one structural concern (only the last batch job returned to the caller for multi-batch local uploads) was flagged in a prior review round and remains unresolved — it means a caller who uploads more than 10 local files in one call silently loses visibility into whether the first N-1 batches actually completed successfully from their end.

nucleus/dataset.py — specifically _upload_local_items_async and the multi-batch job-return strategy.

Important Files Changed

Filename	Overview
nucleus/dataset.py	Core change: append() now always routes to async pipeline. Contains one dead-code guard that can never fire, and delegates multi-batch local uploads to _upload_local_items_async where only the last batch job is returned to the caller (flagged in prior review).
nucleus/dataset_item_uploader.py	Old sync upload path removed; new upload_local_async sends batches as multipart to /append?async=1 and returns all job responses. The redundant item.local guard inside the existence check is harmless since only local items are ever passed here.
nucleus/upload_response.py	File deleted — UploadResponse class removed as promised. Consumers updated in init.py, test_dataset.py, and test_models.py.
tests/test_dataset.py	All append tests migrated to AsyncJob + sleep_until_complete(); old sync assertions removed. New test_dataset_append_rejects_mixed_local_and_remote confirms ValueError on mixed items; test_dataset_append_async_local properly tagged @pytest.mark.integration.
nucleus/init.py	Removed UploadResponse and construct_append_payload exports; populate_dataset deprecated wrapper removed. Clean.

Flowchart

%%{init: {'theme': 'neutral'}}%%
flowchart TD
    A["dataset.append(items, ...)"] --> B{Deprecated params?}
    B -->|asynchronous or batch_size set| C[Emit DeprecationWarning]
    C --> D
    B -->|No| D{What type of items?}
    D -->|LidarScene| E["_append_scenes() → serialize to presigned URL → /upload_scenes?async=1 → AsyncJob"]
    D -->|VideoScene| F["_append_video_scenes() → serialize to presigned URL → /upload_video_scenes?async=1 → AsyncJob"]
    D -->|DatasetItem| G{Empty list?}
    G -->|Yes| H[raise ValueError]
    G -->|No| I{Mixed local + remote?}
    I -->|Yes| J[raise ValueError]
    I -->|Local only| K["_upload_local_items_async() → multipart batches → /append?async=1\nwait for all but last → return last AsyncJob"]
    I -->|Remote only| L["serialize_and_write_to_presigned_url() → /append?async=1 → AsyncJob"]

Loading

%%{init: {'theme': 'base', 'themeVariables': {"darkMode": true, "background": "#0d1117", "primaryColor": "#21262d", "primaryTextColor": "#e6edf3", "primaryBorderColor": "#8b949e", "lineColor": "#8b949e", "textColor": "#e6edf3", "edgeLabelBackground": "#161b22", "actorBkg": "#21262d", "actorBorder": "#8b949e", "actorTextColor": "#e6edf3", "actorLineColor": "#8b949e", "signalColor": "#8b949e", "signalTextColor": "#e6edf3", "noteBkgColor": "#373320", "noteBorderColor": "#d4a72c", "noteTextColor": "#f0e6c0", "labelBoxBkgColor": "#21262d", "labelBoxBorderColor": "#8b949e", "labelTextColor": "#e6edf3", "loopTextColor": "#e6edf3", "activationBkgColor": "#30363d", "activationBorderColor": "#8b949e"}}}%%
flowchart TD
    A["dataset.append(items, ...)"] --> B{Deprecated params?}
    B -->|asynchronous or batch_size set| C[Emit DeprecationWarning]
    C --> D
    B -->|No| D{What type of items?}
    D -->|LidarScene| E["_append_scenes() → serialize to presigned URL → /upload_scenes?async=1 → AsyncJob"]
    D -->|VideoScene| F["_append_video_scenes() → serialize to presigned URL → /upload_video_scenes?async=1 → AsyncJob"]
    D -->|DatasetItem| G{Empty list?}
    G -->|Yes| H[raise ValueError]
    G -->|No| I{Mixed local + remote?}
    I -->|Yes| J[raise ValueError]
    I -->|Local only| K["_upload_local_items_async() → multipart batches → /append?async=1\nwait for all but last → return last AsyncJob"]
    I -->|Remote only| L["serialize_and_write_to_presigned_url() → /append?async=1 → AsyncJob"]

Loading

Comments Outside Diff (1)

1. nucleus/dataset.py, line 2462-2469 (link)

   Only last batch job returned for multi-batch local uploads

   _upload_local_items_async calls uploader.upload_local_async(...) which fires one concurrent request per batch (batch size = local_files_per_upload_request, default 10). Each request creates a separate AsyncJob on the backend. The method then returns AsyncJob.from_json(responses[-1], self._client) — only the last batch's job. If a caller uploads 11+ local items (splitting into ≥ 2 batches), calling sleep_until_complete() on the returned job only waits for the last batch; the earlier batches' jobs are untracked and may still be running. The comment # Return the last job — all batches feed into the same dataset is misleading — they are independent async jobs.

   Prompt To Fix With AI

   This is a comment left during a code review.
   Path: nucleus/dataset.py
   Line: 2462-2469
   
   Comment:
   **Only last batch job returned for multi-batch local uploads**
   
   `_upload_local_items_async` calls `uploader.upload_local_async(...)` which fires one concurrent request per batch (batch size = `local_files_per_upload_request`, default 10). Each request creates a separate `AsyncJob` on the backend. The method then returns `AsyncJob.from_json(responses[-1], self._client)` — only the **last** batch's job. If a caller uploads 11+ local items (splitting into ≥ 2 batches), calling `sleep_until_complete()` on the returned job only waits for the last batch; the earlier batches' jobs are untracked and may still be running. The comment `# Return the last job — all batches feed into the same dataset` is misleading — they are independent async jobs.
   
   How can I resolve this? If you propose a fix, please make it concise.

   

_{Reviews (3): Last reviewed commit: "Address greptile" | 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.