Fix Issue #43211: Make max_concurrency optional (sync only) and fix create_database overloads (#43269)

Copilot · bambriz · allenkim0129 · web-flow · commit 7e59645b2d38 · 2025-10-09T00:31:29.000-04:00
* Initial plan

* Implement max_concurrency optional and fix database method overloads

- Make max_concurrency Optional[int] in read_items (container.py)
- Update default to None to use Python's ThreadPoolExecutor default
- Update _read_items_helper.py to accept Optional[int] for max_concurrency
- Update _cosmos_client_connection.py to default to None
- Fix create_database overloads (sync): remove populate_query_metrics, make offer_throughput keyword-only, add response_hook
- Fix create_database_if_not_exists overloads (sync): same changes
- Fix create_database overloads (async): add response_hook parameter
- Fix create_database_if_not_exists overloads (async): add response_hook parameter
- Update implementations to handle response_hook and remove positional offer_throughput/populate_query_metrics

Co-authored-by: bambriz &lt;8497145+bambriz@users.noreply.github.com&gt;

* Add unit tests for max_concurrency optional behavior

Co-authored-by: bambriz &lt;8497145+bambriz@users.noreply.github.com&gt;

* updated docstrings for create_Database

Updated docstrings for create_Database in sync and async versions.

* 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 &lt;5966120+allenkim0129@users.noreply.github.com&gt;

* Revert async changes - only sync path should be modified

- Revert aio/_container.py to keep max_concurrency: int = 10
- Revert aio/_cosmos_client_connection_async.py to keep default of 10
- Async uses asyncio.Semaphore instead of ThreadPoolExecutor
- Only sync path uses ThreadPoolExecutor and benefits from Optional behavior

Co-authored-by: simorenoh &lt;30335873+simorenoh@users.noreply.github.com&gt;

* update changes

* max concurrency update

updated default of async max currency to be None in public facing api, internally we set to 5.

* pylint and test fixes

* Update _container.py

fix docstring indentation

* update quick fixes

---------

Co-authored-by: copilot-swe-agent[bot] &lt;198982749+Copilot@users.noreply.github.com&gt;
Co-authored-by: bambriz &lt;8497145+bambriz@users.noreply.github.com&gt;
Co-authored-by: bambriz &lt;beambriz2017@gmail.com&gt;
Co-authored-by: allenkim0129 &lt;5966120+allenkim0129@users.noreply.github.com&gt;
Co-authored-by: simorenoh &lt;30335873+simorenoh@users.noreply.github.com&gt;
Co-authored-by: bambriz &lt;bryanambriz@microsoft.com&gt;
diff --git a/sdk/cosmos/azure-cosmos/azure/cosmos/_cosmos_client_connection.py b/sdk/cosmos/azure-cosmos/azure/cosmos/_cosmos_client_connection.py
@@ -1078,7 +1078,7 @@ 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 = ReadItemsHelperSync(
             client=self,
diff --git a/sdk/cosmos/azure-cosmos/azure/cosmos/_read_items_helper.py b/sdk/cosmos/azure-cosmos/azure/cosmos/_read_items_helper.py
@@ -48,7 +48,7 @@ def __init__(
             partition_key_definition: dict[str, Any],
             *,
             executor: Optional[ThreadPoolExecutor] = None,
-            max_concurrency: int = 10,
+            max_concurrency: Optional[int] = None,
             **kwargs: Any
     ):
         self.client = client
@@ -81,6 +81,7 @@ def read_items(self) -> CosmosList:
         if self.executor is not None:
             return self._execute_with_executor(self.executor, query_chunks)
 
+        # Create a new executor; if max_concurrency is None, use ThreadPoolExecutor's default
         with ThreadPoolExecutor(max_workers=self.max_concurrency) as executor:
             return self._execute_with_executor(executor, query_chunks)
 
diff --git a/sdk/cosmos/azure-cosmos/azure/cosmos/aio/_container.py b/sdk/cosmos/azure-cosmos/azure/cosmos/aio/_container.py
@@ -433,7 +433,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,
@@ -449,8 +449,9 @@ 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.
+        :keyword int max_concurrency: Specifies the maximum number of concurrent operations for the
+            `read_items` request. If not provided or set to None, the internal default value will be
+            used when passed to `asyncio.Semaphore`.
         :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.
diff --git a/sdk/cosmos/azure-cosmos/azure/cosmos/aio/_cosmos_client.py b/sdk/cosmos/azure-cosmos/azure/cosmos/aio/_cosmos_client.py
@@ -263,6 +263,7 @@ async def create_database(
         *,
         offer_throughput: Optional[Union[int, ThroughputProperties]] = None,
         initial_headers: Optional[dict[str, str]] = None,
+        response_hook: Optional[Callable[[Mapping[str, Any]], None]] = None,
         throughput_bucket: Optional[int] = None,
         return_properties: Literal[False] = False,
         **kwargs: Any
@@ -271,12 +272,12 @@ async def create_database(
         Create a new database with the given ID (name).
 
         :param str id: ID (name) of the database to create.
-        :keyword offer_throughput: The provisioned throughput for this offer.
-        :paramtype offer_throughput: Union[int, ~azure.cosmos.ThroughputProperties]
+        :keyword Union[int, ~azure.cosmos.ThroughputProperties] offer_throughput: The provisioned throughput
+            for this database.
         :keyword dict[str, str] initial_headers: Initial headers to be sent as part of the request.
-        :keyword response_hook: A callable invoked with the response metadata.
+        :keyword Callable[[dict[str, str], dict[str, Any]], None] response_hook: A callable invoked with
+            the response metadata.
         :keyword int throughput_bucket: The desired throughput bucket for the client
-        :paramtype response_hook: Callable[[dict[str, str], dict[str, Any]], None]
         :keyword bool return_properties: Specifies whether to return either a DatabaseProxy
             or a Tuple containing a DatabaseProxy and the associated database properties.
         :raises ~azure.cosmos.exceptions.CosmosResourceExistsError: Database with the given ID already exists.
@@ -302,6 +303,7 @@ async def create_database(
         *,
         offer_throughput: Optional[Union[int, ThroughputProperties]] = None,
         initial_headers: Optional[dict[str, str]] = None,
+        response_hook: Optional[Callable[[Mapping[str, Any]], None]] = None,
         throughput_bucket: Optional[int] = None,
         return_properties: Literal[True],
         **kwargs: Any
@@ -310,12 +312,12 @@ async def create_database(
         Create a new database with the given ID (name).
 
         :param str id: ID (name) of the database to create.
-        :keyword offer_throughput: The provisioned throughput for this offer.
-        :paramtype offer_throughput: Union[int, ~azure.cosmos.ThroughputProperties]
+        :keyword Union[int, ~azure.cosmos.ThroughputProperties] offer_throughput: The provisioned throughput
+            for this database.
         :keyword dict[str, str] initial_headers: Initial headers to be sent as part of the request.
-        :keyword response_hook: A callable invoked with the response metadata.
+        :keyword Callable[[dict[str, str], dict[str, Any]], None] response_hook: A callable invoked with
+            the response metadata.
         :keyword int throughput_bucket: The desired throughput bucket for the client
-        :paramtype response_hook: Callable[[dict[str, str], dict[str, Any]], None]
         :keyword bool return_properties: Specifies whether to return either a DatabaseProxy
             or a Tuple containing a DatabaseProxy and the associated database properties.
         :raises ~azure.cosmos.exceptions.CosmosResourceExistsError: Database with the given ID already exists.
@@ -345,12 +347,12 @@ async def create_database( # pylint:disable=docstring-should-be-keyword
 
         :param Any args: args
         :param str id: ID (name) of the database to read or create.
-        :keyword offer_throughput: The provisioned throughput for this offer.
-        :paramtype offer_throughput: Union[int, ~azure.cosmos.ThroughputProperties]
+        :keyword Union[int, ~azure.cosmos.ThroughputProperties] offer_throughput: The provisioned throughput
+            for this database.
         :keyword dict[str, str] initial_headers: Initial headers to be sent as part of the request.
-        :keyword response_hook: A callable invoked with the response metadata.
+        :keyword Callable[[dict[str, str], dict[str, Any]], None] response_hook: A callable invoked with
+            the response metadata.
         :keyword int throughput_bucket: The desired throughput bucket for the client
-        :paramtype response_hook: Callable[[dict[str, str], dict[str, Any]], None]
         :keyword bool return_properties: Specifies whether to return either a DatabaseProxy
             or a Tuple containing a DatabaseProxy and the associated database properties.
         :raises ~azure.cosmos.exceptions.CosmosResourceExistsError: Database with the given ID already exists.
@@ -392,11 +394,14 @@ async def create_database( # pylint:disable=docstring-should-be-keyword
 
         offer_throughput = kwargs.pop("offer_throughput", None)
         return_properties = kwargs.pop("return_properties", False)
+        response_hook = kwargs.pop("response_hook", None)
 
         request_options = _build_options(kwargs)
         _set_throughput_options(offer=offer_throughput, request_options=request_options)
 
         result = await self.client_connection.CreateDatabase(database={"id": id}, options=request_options, **kwargs)
+        if response_hook:
+            response_hook(self.client_connection.last_response_headers)
         if not return_properties:
             return DatabaseProxy(self.client_connection, id=result["id"], properties=result)
         return  DatabaseProxy(self.client_connection, id=result["id"], properties=result), result
@@ -408,6 +413,7 @@ async def create_database_if_not_exists(  # pylint: disable=redefined-builtin
         *,
         offer_throughput: Optional[Union[int, ThroughputProperties]] = None,
         initial_headers: Optional[dict[str, str]] = None,
+        response_hook: Optional[Callable[[Mapping[str, Any]], None]] = None,
         throughput_bucket: Optional[int] = None,
         return_properties: Literal[False] = False,
         **kwargs: Any
@@ -422,12 +428,12 @@ async def create_database_if_not_exists(  # pylint: disable=redefined-builtin
             offer throughput if they differ from what is passed in.
 
         :param str id: ID (name) of the database to read or create.
-        :keyword offer_throughput: The provisioned throughput for this offer.
-        :paramtype offer_throughput: Union[int, ~azure.cosmos.ThroughputProperties]
+        :keyword Union[int, ~azure.cosmos.ThroughputProperties] offer_throughput: The provisioned throughput
+            for this database.
         :keyword dict[str, str] initial_headers: Initial headers to be sent as part of the request.
-        :keyword response_hook: A callable invoked with the response metadata.
+        :keyword Callable[[dict[str, str], dict[str, Any]], None] response_hook: A callable invoked with
+            the response metadata.
         :keyword int throughput_bucket: The desired throughput bucket for the client
-        :paramtype response_hook: Callable[[dict[str, str], dict[str, Any]], None]
         :keyword bool return_properties: Specifies whether to return either a DatabaseProxy
             or a Tuple containing a DatabaseProxy and the associated database properties.
         :raises ~azure.cosmos.exceptions.CosmosHttpResponseError: The database read or creation failed.
@@ -443,6 +449,7 @@ async def create_database_if_not_exists(  # pylint: disable=redefined-builtin
         *,
         offer_throughput: Optional[Union[int, ThroughputProperties]] = None,
         initial_headers: Optional[dict[str, str]] = None,
+        response_hook: Optional[Callable[[Mapping[str, Any]], None]] = None,
         throughput_bucket: Optional[int] = None,
         return_properties: Literal[True],
         **kwargs: Any
@@ -457,12 +464,12 @@ async def create_database_if_not_exists(  # pylint: disable=redefined-builtin
             offer throughput if they differ from what is passed in.
 
         :param str id: ID (name) of the database to read or create.
-        :keyword offer_throughput: The provisioned throughput for this offer.
-        :paramtype offer_throughput: Union[int, ~azure.cosmos.ThroughputProperties]
+        :keyword Union[int, ~azure.cosmos.ThroughputProperties] offer_throughput: The provisioned throughput
+            for this database.
         :keyword dict[str, str] initial_headers: Initial headers to be sent as part of the request.
-        :keyword response_hook: A callable invoked with the response metadata.
+        :keyword Callable[[dict[str, str], dict[str, Any]], None] response_hook: A callable invoked with
+            the response metadata.
         :keyword int throughput_bucket: The desired throughput bucket for the client
-        :paramtype response_hook: Callable[[dict[str, str], dict[str, Any]], None]
         :keyword bool return_properties: Specifies whether to return either a DatabaseProxy
             or a Tuple containing a DatabaseProxy and the associated database properties.
         :raises ~azure.cosmos.exceptions.CosmosHttpResponseError: The database read or creation failed.
@@ -488,12 +495,12 @@ async def create_database_if_not_exists( # pylint:disable=docstring-should-be-ke
 
         :param Any args: args
         :param str id: ID (name) of the database to read or create.
-        :keyword offer_throughput: The provisioned throughput for this offer.
-        :paramtype offer_throughput: Union[int, ~azure.cosmos.ThroughputProperties]
+        :keyword Union[int, ~azure.cosmos.ThroughputProperties] offer_throughput: The provisioned throughput
+            for this database.
         :keyword dict[str, str] initial_headers: Initial headers to be sent as part of the request.
-        :keyword response_hook: A callable invoked with the response metadata.
+        :keyword Callable[[dict[str, str], dict[str, Any]], None] response_hook: A callable invoked with
+            the response metadata.
         :keyword int throughput_bucket: The desired throughput bucket for the client
-        :paramtype response_hook: Callable[[dict[str, str], dict[str, Any]], None]
         :keyword bool return_properties: Specifies whether to return either a DatabaseProxy
             or a Tuple containing a DatabaseProxy and the associated database properties.
         :raises ~azure.cosmos.exceptions.CosmosHttpResponseError: The database read or creation failed.
diff --git a/sdk/cosmos/azure-cosmos/azure/cosmos/aio/_cosmos_client_connection_async.py b/sdk/cosmos/azure-cosmos/azure/cosmos/aio/_cosmos_client_connection_async.py
@@ -2284,7 +2284,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,
diff --git a/sdk/cosmos/azure-cosmos/azure/cosmos/aio/_read_items_helper_async.py b/sdk/cosmos/azure-cosmos/azure/cosmos/aio/_read_items_helper_async.py
@@ -48,7 +48,7 @@ def __init__(
             items: Sequence[Tuple[str, PartitionKeyType]],
             options: Optional[Mapping[str, Any]],
             partition_key_definition: dict[str, Any],
-            max_concurrency: int = 10,
+            max_concurrency: int = 5,
             **kwargs: Any
     ):
         self.client = client
@@ -57,7 +57,7 @@ def __init__(
         self.options = dict(options) if options is not None else {}
         self.partition_key_definition = partition_key_definition
         self.kwargs = kwargs
-        self.max_concurrency = max_concurrency
+        self.max_concurrency = max_concurrency if max_concurrency and max_concurrency > 0 else 5
         self.max_items_per_query = 1000
 
     async def read_items(self) -> 'CosmosList':
diff --git a/sdk/cosmos/azure-cosmos/azure/cosmos/container.py b/sdk/cosmos/azure-cosmos/azure/cosmos/container.py
@@ -292,7 +292,7 @@ def read_items(
             items: Sequence[Tuple[str, PartitionKeyType]],
             *,
             executor: Optional[ThreadPoolExecutor] = None,
-            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,
@@ -311,7 +311,8 @@ def read_items(
         :keyword executor: Optional ThreadPoolExecutor for handling concurrent operations.
                       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. Defaults to 10.
+                      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.
         :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.
diff --git a/sdk/cosmos/azure-cosmos/azure/cosmos/cosmos_client.py b/sdk/cosmos/azure-cosmos/azure/cosmos/cosmos_client.py
