feat: async + multi-source create_benchmark

CHANGELOG.md

@@ -5,6 +5,14 @@ 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.20.0](https://github.com/scaleapi/nucleus-python-client/releases/tag/v0.20.0) - 2026-07-27
+
+### Added
+- **Multi-source `create_benchmark()`.** Members can now come from any combination of `item_ids`, `(dataset_id, ref_id)` `items`, one or more slices (`slice_id` / `slice_ids`), and one or more datasets (`dataset_id` / `dataset_ids`) — unioned and de-duplicated server-side. At least one source is required (previously exactly one).
+
+### Changed
+- **`create_benchmark()` is now asynchronous.** The server creates the benchmark in a `"building"` state and streams its members in via a background job (removing the previous item-count ceiling on slice/dataset-sourced benchmarks). `create_benchmark()` blocks on that job by default and returns the completed `"ready"` benchmark — the return type is unchanged, so existing blocking callers are unaffected. Pass `wait_for_completion=False` to return the `"building"` benchmark immediately and poll it yourself via `Benchmark.refresh()` (checking the new `Benchmark.status` field). A failed build job raises `JobError`. `Benchmark` now exposes `status` (`"building"` / `"ready"` / `"failed"`).
+
 ## [0.19.0](https://github.com/scaleapi/nucleus-python-client/releases/tag/v0.19.0) - 2026-07-10
 
 ### Added

nucleus/__init__.py

@@ -1096,15 +1096,26 @@ def create_benchmark(
         items: Optional[List[Dict[str, str]]] = None,
         slice_id: Optional[str] = None,
         dataset_id: Optional[str] = None,
+        slice_ids: Optional[List[str]] = None,
+        dataset_ids: Optional[List[str]] = None,
+        wait_for_completion: bool = True,
+        verbose: bool = True,
     ) -> Benchmark:
         """Create a benchmark from ground-truth items.
 
-        Provide the members through exactly one source: explicit ``item_ids``,
-        ``(dataset_id, ref_id)`` pairs via ``items``, all items in a slice via
-        ``slice_id``, or all items in a dataset via ``dataset_id``. Items
-        without ground truth are skipped (reported on the returned benchmark
-        as ``skipped_items_without_ground_truth``). Membership is frozen at
-        creation.
+        Provide members through any combination of sources: explicit
+        ``item_ids``, ``(dataset_id, ref_id)`` pairs via ``items``, one or more
+        slices via ``slice_id`` / ``slice_ids``, and one or more datasets via
+        ``dataset_id`` / ``dataset_ids``. Members are unioned and de-duplicated
+        across all sources; at least one source is required. Items without
+        ground truth are skipped. Membership is frozen at creation.
+
+        Creation is **asynchronous**: the server creates the benchmark in a
+        ``"building"`` state and streams its members in via a background job.
+        By default this method blocks until that job finishes and returns the
+        completed (``"ready"``) benchmark. Pass ``wait_for_completion=False``
+        to return immediately with a ``"building"`` benchmark you can poll via
+        :meth:`Benchmark.refresh` (checking ``benchmark.status``).
 
         Parameters:
             name: Benchmark display name.
@@ -1114,19 +1125,32 @@ def create_benchmark(
             items: ``{"dataset_id": ..., "ref_id": ...}`` pairs.
             slice_id: Slice id (``slc_*``) whose items become members.
             dataset_id: Dataset id (``ds_*``) whose items become members.
+            slice_ids: Multiple slice ids whose items become members.
+            dataset_ids: Multiple dataset ids whose items become members.
+            wait_for_completion: Block until the build job finishes and return
+                the ready benchmark (default). If ``False``, return the
+                ``"building"`` benchmark immediately.
+            verbose: Log build-job polling progress while waiting.
 
         Returns:
-            :class:`Benchmark`: The created benchmark.
+            :class:`Benchmark`: The created benchmark — ``"ready"`` when
+            ``wait_for_completion`` is ``True``, otherwise ``"building"``.
         """
-        sources = [
+        has_source = any(
             source
-            for source in (item_ids, items, slice_id, dataset_id)
-            if source is not None
-        ]
-        if len(sources) != 1:
+            for source in (
+                item_ids,
+                items,
+                slice_id,
+                dataset_id,
+                slice_ids,
+                dataset_ids,
+            )
+        )
+        if not has_source:
             raise ValueError(
-                "Provide exactly one of item_ids, items, slice_id, or "
-                "dataset_id to define benchmark membership"
+                "Provide at least one of item_ids, items, slice_id(s), or "
+                "dataset_id(s) to define benchmark membership"
             )
         payload: Dict[str, Any] = {"name": name}
         if description is not None:
@@ -1141,8 +1165,26 @@ def create_benchmark(
             payload["slice_id"] = slice_id
         if dataset_id is not None:
             payload["dataset_id"] = dataset_id
-        data = self.post(payload, "benchmarks")
-        return Benchmark.from_json(data, self)
+        if slice_ids is not None:
+            payload["slice_ids"] = slice_ids
+        if dataset_ids is not None:
+            payload["dataset_ids"] = dataset_ids
+
+        # Async: the server responds 202 with {benchmark_id, job_id}. The
+        # benchmark row already exists (in 'building' state); the build job
+        # streams members in and flips it to 'ready' (or 'failed').
+        response = self.post(payload, "benchmarks")
+        benchmark_id = response["benchmark_id"]
+        if wait_for_completion:
+            if job_id is None:
+                raise ValueError(
+                    "Server did not return a job_id in the create-benchmark "
+                    "response; cannot poll for completion. Pass "
+                    "wait_for_completion=False to suppress this error."
+                )
+            self.get_job(job_id).sleep_until_complete(verbose_std_out=verbose)
+
+        return self.get_benchmark(benchmark_id)
 
     def list_benchmarks(self) -> List[Benchmark]:
         """List benchmarks visible to the current user.

nucleus/benchmark.py

@@ -46,6 +46,9 @@ class Benchmark:
     item_count: Optional[int] = None
     dataset_count: Optional[int] = None
     skipped_items_without_ground_truth: Optional[int] = None
+    #: Async build lifecycle: ``"building"`` until the server's build job finishes
+    #: streaming members in, then ``"ready"`` (or ``"failed"``).
+    status: Optional[str] = None
     _client: Optional["NucleusClient"] = field(repr=False, default=None)
 
     @classmethod
@@ -66,6 +69,7 @@ def from_json(
             skipped_items_without_ground_truth=payload.get(
                 "skipped_items_without_ground_truth"
             ),
+            status=payload.get("status"),
             _client=client,
         )
 

pyproject.toml

@@ -25,7 +25,7 @@ ignore = ["E501", "E741", "E731", "F401"]  # Easy ignore for getting it running
 
 [tool.poetry]
 name = "scale-nucleus"
-version = "0.19.0"
+version = "0.20.0"
 description = "The official Python client library for Nucleus, the Data Platform for AI"
 license =  "MIT"
 authors = ["Scale AI Nucleus Team <nucleusapi@scaleapi.com>"]

tests/test_benchmarks.py

@@ -48,9 +48,31 @@ def test_benchmark_from_json_maps_benchmark_id():
     assert b.skipped_items_without_ground_truth == 3
 
 
-def test_create_benchmark_from_slice():
-    client = NucleusClient(api_key="test")
-    client.connection.post = MagicMock(return_value=dict(_BENCHMARK_ROW))
+def test_benchmark_from_json_parses_status():
+    assert Benchmark.from_json(_BENCHMARK_ROW).status is None
+    assert (
+        Benchmark.from_json({**_BENCHMARK_ROW, "status": "building"}).status
+        == "building"
+    )
+
+
+def _mock_async_create(client, *, benchmark_row=None):
+    """Wire up the async create_benchmark flow: 202 {benchmark_id, job_id} ->
+    poll the build job -> re-fetch the ready benchmark."""
+    client.connection.post = MagicMock(
+        return_value={"benchmark_id": "bm_1", "job_id": "job_1"}
+    )
+    client.get_job = MagicMock()  # .sleep_until_complete() is a no-op MagicMock
+    client.get_benchmark = MagicMock(
+        return_value=Benchmark.from_json(
+            benchmark_row or {**_BENCHMARK_ROW, "status": "ready"}, client
+        )
+    )
+    return client
+
+
+def test_create_benchmark_from_slice_polls_then_returns_ready():
+    client = _mock_async_create(NucleusClient(api_key="test"))
     benchmark = client.create_benchmark(
         "city-streets", description="desc", slice_id="slc_1"
     )
@@ -61,12 +83,16 @@ def test_create_benchmark_from_slice():
         "description": "desc",
         "slice_id": "slc_1",
     }
+    # Blocks on the build job by default, then fetches the ready benchmark.
+    client.get_job.assert_called_once_with("job_1")
+    client.get_job.return_value.sleep_until_complete.assert_called_once()
+    client.get_benchmark.assert_called_once_with("bm_1")
     assert benchmark.id == "bm_1"
+    assert benchmark.status == "ready"
 
 
 def test_create_benchmark_from_item_ids_and_metadata():
-    client = NucleusClient(api_key="test")
-    client.connection.post = MagicMock(return_value=dict(_BENCHMARK_ROW))
+    client = _mock_async_create(NucleusClient(api_key="test"))
     client.create_benchmark(
         "city-streets",
         metadata={"team": "av"},
@@ -77,12 +103,39 @@ def test_create_benchmark_from_item_ids_and_metadata():
     assert payload["metadata"] == {"team": "av"}
 
 
-def test_create_benchmark_requires_exactly_one_member_source():
+def test_create_benchmark_no_wait_returns_building_without_polling():
+    client = _mock_async_create(
+        NucleusClient(api_key="test"),
+        benchmark_row={**_BENCHMARK_ROW, "status": "building", "item_count": 0},
+    )
+    benchmark = client.create_benchmark(
+        "city-streets", slice_id="slc_1", wait_for_completion=False
+    )
+    client.get_job.assert_not_called()
+    client.get_benchmark.assert_called_once_with("bm_1")
+    assert benchmark.status == "building"
+
+
+def test_create_benchmark_requires_at_least_one_member_source():
     client = NucleusClient(api_key="test")
-    with pytest.raises(ValueError, match="exactly one"):
+    with pytest.raises(ValueError, match="at least one"):
         client.create_benchmark("b")
-    with pytest.raises(ValueError, match="exactly one"):
-        client.create_benchmark("b", slice_id="slc_1", dataset_id="ds_1")
+    with pytest.raises(ValueError, match="at least one"):
+        client.create_benchmark("b", slice_ids=[], dataset_ids=[])
+
+
+def test_create_benchmark_combines_multiple_sources():
+    client = _mock_async_create(NucleusClient(api_key="test"))
+    client.create_benchmark(
+        "multi",
+        item_ids=["di_3"],
+        slice_ids=["slc_1", "slc_2"],
+        dataset_ids=["ds_1", "ds_2"],
+    )
+    payload = client.connection.post.call_args[0][0]
+    assert payload["item_ids"] == ["di_3"]
+    assert payload["slice_ids"] == ["slc_1", "slc_2"]
+    assert payload["dataset_ids"] == ["ds_1", "ds_2"]
 
 
 def test_list_benchmarks():