feat: async + multi-source create_benchmark

CHANGELOG.md

@@ -5,6 +5,11 @@ 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
+
+### 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,22 @@ def create_benchmark(
         items: Optional[List[Dict[str, str]]] = None,
         slice_id: Optional[str] = None,
         dataset_id: Optional[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.
+        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,9 +1121,14 @@ 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.
+            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 = [
             source
@@ -1141,8 +1153,18 @@ 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)
+
+        # 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"]
+        job_id = response.get("job_id")
+
+        if wait_for_completion and job_id is not None:
+            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,6 +103,19 @@ def test_create_benchmark_from_item_ids_and_metadata():
     assert payload["metadata"] == {"team": "av"}
 
 
+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_exactly_one_member_source():
     client = NucleusClient(api_key="test")
     with pytest.raises(ValueError, match="exactly one"):