diff --git a/src/bloomy/utils/async_base_operations.py b/src/bloomy/utils/async_base_operations.py index fb2f10d..ad1a772 100644 --- a/src/bloomy/utils/async_base_operations.py +++ b/src/bloomy/utils/async_base_operations.py @@ -93,29 +93,25 @@ async def _process_bulk_async[T]( async def create_single( index: int, item_data: dict[str, Any] - ) -> tuple[int, T | BulkCreateError]: + ) -> T | BulkCreateError: async with semaphore: try: self._validate_bulk_item(item_data, required_fields) - created = await create_func(item_data) - return (index, created) + return await create_func(item_data) except Exception as e: - error = BulkCreateError( + return BulkCreateError( index=index, input_data=item_data, error=str(e) ) - return (index, error) - tasks = [ - create_single(index, item_data) for index, item_data in enumerate(items) - ] - results = await asyncio.gather(*tasks) - results_list = list(results) - results_list.sort(key=lambda x: x[0]) + # asyncio.gather preserves input order, so no post-sort is needed. + results = await asyncio.gather( + *(create_single(index, item_data) for index, item_data in enumerate(items)) + ) successful: list[T] = [] failed: list[BulkCreateError] = [] - for _, result in results_list: + for result in results: if isinstance(result, BulkCreateError): failed.append(result) else: diff --git a/tests/test_async_base_operations.py b/tests/test_async_base_operations.py index 8ca7d5c..78b93bb 100644 --- a/tests/test_async_base_operations.py +++ b/tests/test_async_base_operations.py @@ -81,3 +81,50 @@ async def test_get_default_user_id(self) -> None: # Verify API call client.get.assert_called_once_with("users/mine") + + @pytest.mark.asyncio + async def test_process_bulk_async_preserves_input_order(self) -> None: + """Gather results stay in input order even when later items finish first.""" + import asyncio + + client = MockAsyncHTTPClient() + ops = AsyncBaseOperations(client) + + # Delays inverted so item 0 finishes last, item 2 finishes first. + delays = {0: 0.05, 1: 0.02, 2: 0.01} + + async def create_func(item_data: dict) -> str: + await asyncio.sleep(delays[item_data["index"]]) + return f"created-{item_data['index']}" + + items = [{"index": i, "title": f"item-{i}"} for i in range(3)] + result = await ops._process_bulk_async( + items, create_func, required_fields=["title"], max_concurrent=3 + ) + + assert result.failed == [] + assert result.successful == ["created-0", "created-1", "created-2"] + + @pytest.mark.asyncio + async def test_process_bulk_async_failure_index_matches_input(self) -> None: + """Failed items keep the original input index without post-sort.""" + import asyncio + + client = MockAsyncHTTPClient() + ops = AsyncBaseOperations(client) + + async def create_func(item_data: dict) -> str: + await asyncio.sleep(0.01 if item_data["index"] != 0 else 0.03) + if item_data["index"] == 1: + raise ValueError("boom") + return f"created-{item_data['index']}" + + items = [{"index": i, "title": f"item-{i}"} for i in range(3)] + result = await ops._process_bulk_async( + items, create_func, required_fields=["title"], max_concurrent=3 + ) + + assert result.successful == ["created-0", "created-2"] + assert len(result.failed) == 1 + assert result.failed[0].index == 1 + assert "boom" in result.failed[0].error