feat(dataset): let get_items choose its thread count

get_items() read at a fixed 4 threads, so reaching the faster end of the
range meant flattening stream_items() by hand. Accept num_threads and forward
it. Appended to the signature, so existing positional callers are unaffected;
validation is inherited from stream_items().

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

Author: alexkuzmik

apps/opik-documentation/documentation/fern/docs-v2/evaluation/advanced/manage_datasets.mdx

@@ -415,33 +415,58 @@ You can download a dataset from Opik using the `get_dataset` method:
   ```
 </CodeBlocks>
 
-### Downloading large datasets
+### Downloading large datasets faster
 
-`get_items()` returns the whole dataset as a single list, so the call does not return until every
-item has been downloaded and the full result is held in memory.
+Dataset items are fetched a page at a time, and those pages are downloaded concurrently. Raise
+`num_threads` to speed up a large read — the default is 4, and 8 is usually worth trying on
+datasets of tens of thousands of items:
+
+```python title="Python" language="python"
+from opik import Opik
+
+client = Opik()
+dataset = client.get_dataset(name="My dataset")
 
-`stream_items()` reads the same items in chunks instead, yielding each chunk as soon as it
-arrives. Use it when you want to start processing before the download finishes, or when the
-dataset is too large to hold in memory all at once:
+# The whole dataset as one list, downloaded over 8 threads
+items = dataset.get_items(num_threads=8)
+```
+
+The thread count never changes the result, only how quickly it arrives.
+
+`get_items()` returns the whole dataset as a single list, so the call does not return until every
+item has been downloaded and the full result is held in memory. `stream_items()` reads the same
+items in chunks instead, yielding each chunk as soon as it arrives. Use it when you want to start
+processing before the download finishes, or when the dataset is too large to hold in memory all at
+once:
 
 ```python title="Python" language="python"
 from opik import Opik
 
 client = Opik()
 dataset = client.get_dataset(name="My dataset")
 
-for chunk in dataset.stream_items(num_threads=8):
-    process(chunk)  # chunk is a list of dicts, exactly as get_items() returns them
+# One chunk at a time, instead of the whole dataset at once
+for chunk in dataset.stream_items(chunk_size=5000, num_threads=8):
+    process(chunk)  # a list of dicts, exactly as get_items() returns them
 ```
 
-It accepts the same `filter_string` as `get_items()`, and `nb_samples` to stop after a given
-number of items. Chunks arrive in dataset order; only the last one may be shorter than
-`chunk_size`.
+Chunks arrive in dataset order; only the last one may be shorter than `chunk_size`. Both methods
+accept the same `filter_string`, and `nb_samples` to stop after a given number of items:
+
+```python title="Python" language="python"
+for chunk in dataset.stream_items(
+    num_threads=8,
+    filter_string='data.category = "geography"',
+    nb_samples=10_000,
+):
+    process(chunk)
+```
 
 <Note>
-  Both methods fetch pages concurrently — `get_items()` is built on `stream_items()` — so raising
-  `num_threads` speeds up either read. The default of 4 is a safe starting point; 8 is usually
-  worth trying on datasets of tens of thousands of items.
+  `chunk_size` controls how many items each request fetches (default 2000). Fetching a chunk costs
+  a fixed overhead whatever its size, so lowering it makes the whole read slower — raise it to
+  trade memory for speed. Lower it only when individual items are large, since up to
+  `2 * num_threads` chunks are held in memory at once.
 </Note>
 
 ## Filtering datasets programmatically

sdks/python/src/opik/api_objects/dataset/dataset.py

@@ -121,12 +121,20 @@ def get_items(
         self,
         nb_samples: Optional[int] = None,
         filter_string: Optional[str] = None,
+        num_threads: int = constants.DATASET_ITEMS_READ_NUM_THREADS,
     ) -> List[Dict[str, Any]]:
         """
         Retrieve dataset items as a list of dictionaries.
 
         Args:
             nb_samples: Maximum number of items to retrieve. If not set, all items are returned.
+            num_threads: Number of item pages fetched concurrently. Must be a
+                positive integer, defaults to 4; pass ``1`` to fetch
+                sequentially. Raising it speeds up large reads at the cost of
+                more load on the backend. Capped at
+                ``constants.DATASET_ITEMS_READ_MAX_THREADS``. Use
+                :meth:`stream_items` instead when the dataset is too large to
+                hold in memory all at once.
             filter_string: Optional OQL filter string to filter dataset items.
                 Supports filtering by tags, data fields, metadata, etc.
 
@@ -144,11 +152,17 @@ def get_items(
 
         Returns:
             A list of dictionaries representing the dataset items.
+
+        Raises:
+            ValueError: If ``num_threads`` is not a positive integer, or
+                ``nb_samples`` is not a positive integer.
         """
         return [
             item
             for chunk in self.stream_items(
-                filter_string=filter_string, nb_samples=nb_samples
+                filter_string=filter_string,
+                nb_samples=nb_samples,
+                num_threads=num_threads,
             )
             for item in chunk
         ]

sdks/python/tests/unit/api_objects/dataset/test_stream_item_chunks.py

@@ -385,3 +385,48 @@ def test_get_items__empty_dataset__returns_empty_list():
     dataset = _build_dataset(endpoint)
 
     assert dataset.get_items() == []
+
+
+def test_get_items__num_threads__forwarded_to_the_reader():
+    chunk = constants.DATASET_STREAM_BATCH_SIZE
+    endpoint = FakeItemsEndpoint(_rest_items(chunk * 4), delay_seconds=0.05)
+    dataset = _build_dataset(endpoint)
+
+    items = dataset.get_items(num_threads=4)
+
+    assert len(items) == chunk * 4
+    assert endpoint.max_in_flight > 1
+
+
+def test_get_items__num_threads_one__fetches_pages_sequentially():
+    chunk = constants.DATASET_STREAM_BATCH_SIZE
+    endpoint = FakeItemsEndpoint(_rest_items(chunk * 3), delay_seconds=0.05)
+    dataset = _build_dataset(endpoint)
+
+    items = dataset.get_items(num_threads=1)
+
+    assert len(items) == chunk * 3
+    assert endpoint.max_in_flight == 1
+
+
+def test_get_items__thread_count_does_not_change_the_result():
+    endpoint = FakeItemsEndpoint(_rest_items(constants.DATASET_STREAM_BATCH_SIZE * 3))
+    dataset = _build_dataset(endpoint)
+
+    assert dataset.get_items(num_threads=1) == dataset.get_items(num_threads=8)
+
+
+@pytest.mark.parametrize("num_threads", [0, -1, True, "4"])
+def test_get_items__invalid_num_threads__raises_value_error(num_threads):
+    dataset = _build_dataset(endpoint=None)
+
+    with pytest.raises(ValueError):
+        dataset.get_items(num_threads=num_threads)
+
+
+def test_get_items__positional_args_unchanged__nb_samples_still_first():
+    """num_threads was appended, so existing positional callers are unaffected."""
+    endpoint = FakeItemsEndpoint(_rest_items(50))
+    dataset = _build_dataset(endpoint)
+
+    assert len(dataset.get_items(10)) == 10