Address review comments: update docstrings, add tests, fix async version

- Add max_concurrency to docstring in _cosmos_client_connection.py
- Update docstring to use generic language about ThreadPoolExecutor default
- Add tests for max_concurrency ignored with executor and Python default behavior
- Update async version of read_items to also use Optional[int] for max_concurrency
- Update async _cosmos_client_connection_async.py to use None as default

Co-authored-by: allenkim0129 <[email protected]>

Author: Copilot

sdk/cosmos/azure-cosmos/azure/cosmos/_cosmos_client_connection.py

@@ -1068,6 +1068,9 @@ def read_items(
         :param dict options: The request options for the request.
         :keyword executor: Optional ThreadPoolExecutor for thread management
         :paramtype executor: Optional[ThreadPoolExecutor]
+        :keyword int max_concurrency: The maximum number of concurrent operations for the items request.
+            This value is ignored if an executor is provided. If not specified, the default max_concurrency
+            defined by Python's ThreadPoolExecutor will be applied.
         :return: The list of read items.
         :rtype: CosmosList
         """

sdk/cosmos/azure-cosmos/azure/cosmos/aio/_container.py

@@ -437,7 +437,7 @@ async def read_items(
             self,
             items: Sequence[Tuple[str, PartitionKeyType]],
             *,
-            max_concurrency: int = 10,
+            max_concurrency: Optional[int] = None,
             consistency_level: Optional[str] = None,
             session_token: Optional[str] = None,
             initial_headers: Optional[Dict[str, str]] = None,
@@ -454,7 +454,8 @@ async def read_items(
         :param items: A list of tuples, where each tuple contains an item's ID and partition key.
         :type items: Sequence[Tuple[str, PartitionKeyType]]
         :keyword int max_concurrency: The maximum number of concurrent operations for the read_items
-            request. Defaults to 10.
+            request. If not specified, the default max_concurrency defined by Python's ThreadPoolExecutor
+            will be applied.
         :keyword str consistency_level: The consistency level to use for the request.
         :keyword str session_token: Token for use with Session consistency.
         :keyword dict[str, str] initial_headers: Initial headers to be sent as part of the request.

sdk/cosmos/azure-cosmos/azure/cosmos/aio/_cosmos_client_connection_async.py

@@ -2279,6 +2279,8 @@ async def read_items(
         :param items: A list of tuples, where each tuple contains an item's ID and partition key.
         :type items: Sequence[Tuple[str, _PartitionKeyType]]
         :param dict options: The request options for the request.
+        :keyword int max_concurrency: The maximum number of concurrent operations for the items request.
+            If not specified, the default max_concurrency defined by Python's ThreadPoolExecutor will be applied.
         :return: The list of read items.
         :rtype: CosmosList
         """
@@ -2292,7 +2294,7 @@ async def read_items(
             raise ValueError("Could not find partition key definition for collection.")
 
         # Extract and remove max_concurrency from kwargs
-        max_concurrency = kwargs.pop('max_concurrency', 10)
+        max_concurrency = kwargs.pop('max_concurrency', None)
         helper = ReadItemsHelperAsync(
             client=self,
             collection_link=collection_link,

sdk/cosmos/azure-cosmos/azure/cosmos/container.py

@@ -313,7 +313,7 @@ def read_items(
                       If not provided, a new executor will be created as needed.
         :keyword int max_concurrency: The maximum number of concurrent operations for the
                       items request. This value is ignored if an executor is provided. If not specified,
-                      defaults to Python's ThreadPoolExecutor default of min(32, (os.cpu_count() or 1) + 4).
+                      the default max_concurrency defined by Python's ThreadPoolExecutor will be applied.
         :keyword str consistency_level: The consistency level to use for the request.
         :keyword str session_token: Token for use with Session consistency.
         :keyword dict[str, str] initial_headers: Initial headers to be sent as part of the request.

sdk/cosmos/azure-cosmos/tests/test_max_concurrency.py

@@ -1,9 +1,10 @@
 # -*- coding: utf-8 -*-
 # The MIT License (MIT)
 # Copyright (c) Microsoft Corporation. All rights reserved.
+import os
 import unittest
 from concurrent.futures import ThreadPoolExecutor
-from unittest.mock import Mock
+from unittest.mock import Mock, patch
 from azure.cosmos._read_items_helper import ReadItemsHelperSync
 
 
@@ -49,6 +50,66 @@ def test_max_concurrency_value_is_preserved(self):
         # Verify that max_concurrency is set correctly
         self.assertEqual(helper.max_concurrency, 5)
 
+    def test_max_concurrency_ignored_with_executor(self):
+        """Test that max_concurrency is ignored if an executor is provided."""
+        # Mock client connection
+        mock_client = Mock()
+        mock_client._routing_map_provider = Mock()
+        mock_client._routing_map_provider.get_overlapping_ranges = Mock(return_value=[])
+        
+        # Create a mock executor
+        mock_executor = Mock(spec=ThreadPoolExecutor)
+        
+        # Create helper with both executor and max_concurrency
+        helper = ReadItemsHelperSync(
+            client=mock_client,
+            collection_link="/dbs/test/colls/test",
+            items=[],
+            options={},
+            partition_key_definition={"paths": ["/id"], "kind": "Hash"},
+            executor=mock_executor,
+            max_concurrency=5
+        )
+        
+        # Verify that executor is set and max_concurrency is preserved
+        self.assertEqual(helper.executor, mock_executor)
+        self.assertEqual(helper.max_concurrency, 5)
+        
+        # When read_items is called with empty items, it returns early
+        # But we can verify the executor is stored
+        result = helper.read_items()
+        self.assertIsNotNone(result)
+
+    def test_default_max_concurrency_uses_python_default(self):
+        """Test that if max_concurrency not specified, defaults to Python's ThreadPoolExecutor default."""
+        # Get Python's default calculation
+        cpu_count = os.cpu_count() or 1
+        expected_default = min(32, cpu_count + 4)
+        
+        # Create a ThreadPoolExecutor with None to see what Python uses
+        with ThreadPoolExecutor(max_workers=None) as executor:
+            actual_workers = executor._max_workers
+            
+        # Verify the expected default matches Python's calculation
+        self.assertEqual(actual_workers, expected_default)
+        
+        # Now verify that our helper will pass None to ThreadPoolExecutor
+        mock_client = Mock()
+        mock_client._routing_map_provider = Mock()
+        mock_client._routing_map_provider.get_overlapping_ranges = Mock(return_value=[])
+        
+        helper = ReadItemsHelperSync(
+            client=mock_client,
+            collection_link="/dbs/test/colls/test",
+            items=[],
+            options={},
+            partition_key_definition={"paths": ["/id"], "kind": "Hash"},
+            max_concurrency=None
+        )
+        
+        # Verify None is passed through
+        self.assertIsNone(helper.max_concurrency)
+
 
 if __name__ == '__main__':
     unittest.main()