Azure · simorenoh · Nov 21, 2025 · Jun 15, 2025 · Jun 16, 2025 · Jun 16, 2025
@@ -19,6 +19,7 @@
 
 #### Features Added
 * Added feed range support in `query_items`. See [PR 41722](https://github.com/Azure/azure-sdk-for-python/pull/41722).
+* Added support for Per Partition Automatic Failover. To enable this feature, you must follow the guide [here](https://learn.microsoft.com/azure/cosmos-db/how-to-configure-per-partition-automatic-failover). See [PR 41588](https://github.com/Azure/azure-sdk-for-python/pull/41588).
 
 #### Bugs Fixed
 * Fixed session container session token logic. The SDK will now only send the relevant partition-local session tokens for read document requests and write requests when multi-region writes are enabled, as opposed to the entire compound session token for the container for every document request. See [PR 41678](https://github.com/Azure/azure-sdk-for-python/pull/41678).

@@ -940,6 +940,11 @@ requests to another region:
 - `AZURE_COSMOS_FAILURE_PERCENTAGE_TOLERATED`: Default is a `90` percent failure rate.
   - After a partition reaches a 90 percent failure rate for all requests, the SDK will send requests routed to that partition to another region.
 
+### Per Partition Automatic Failover (Public Preview)
+Per partition automatic failover enables the SDK to automatically redirect write requests at the partition level to another region based on service-side signals. This feature is available 
+only for single write region accounts that have at least one read-only region. When per partition automatic failover is enabled, per partition circuit breaker and hedging is enabled by default, meaning 
+all its configurable options also apply to per partition automatic failover. To enable this feature, follow the guide [here](https://learn.microsoft.com/azure/cosmos-db/how-to-configure-per-partition-automatic-failover).
+
 ## Troubleshooting
 
 ### General

@@ -28,6 +28,7 @@
 import uuid
 import re
 import binascii
+import os
 from typing import Dict, Any, List, Mapping, Optional, Sequence, Union, Tuple, TYPE_CHECKING
 
 from urllib.parse import quote as urllib_quote
@@ -45,9 +46,13 @@
 if TYPE_CHECKING:
     from ._cosmos_client_connection import CosmosClientConnection
     from .aio._cosmos_client_connection_async import CosmosClientConnection as AsyncClientConnection
+    from ._global_partition_endpoint_manager_per_partition_automatic_failover import (
+        _GlobalPartitionEndpointManagerForPerPartitionAutomaticFailover)
     from ._request_object import RequestObject
+    from ._routing.routing_range import PartitionKeyRangeWrapper
 
 # pylint: disable=protected-access
+#cspell:ignore PPAF, ppaf
 
 _COMMON_OPTIONS = {
     'initial_headers': 'initialHeaders',
@@ -935,3 +940,22 @@ def _build_properties_cache(properties: Dict[str, Any], container_link: str) ->
         "_self": properties.get("_self", None), "_rid": properties.get("_rid", None),
         "partitionKey": properties.get("partitionKey", None), "container_link": container_link
     }
+
+def try_ppaf_failover_threshold(
+        global_endpoint_manager: "_GlobalPartitionEndpointManagerForPerPartitionAutomaticFailover",
+        pk_range_wrapper: "PartitionKeyRangeWrapper",
+        request: "RequestObject"):
+    # If PPAF is enabled, we track consecutive failures for certain exceptions, and only fail over at a partition
+    # level after the threshold is reached
+    if request and global_endpoint_manager.is_per_partition_automatic_failover_applicable(request):
+        if (global_endpoint_manager.ppaf_thresholds_tracker.get_pk_failures(pk_range_wrapper)
+                >= int(os.environ.get(Constants.TIMEOUT_ERROR_THRESHOLD_PPAF,
+                                      Constants.TIMEOUT_ERROR_THRESHOLD_PPAF_DEFAULT))):
+            # If the PPAF threshold is reached, we reset the count and retry to the next region
+            global_endpoint_manager.ppaf_thresholds_tracker.clear_pk_failures(pk_range_wrapper)
+            partition_level_info = global_endpoint_manager.partition_range_to_failover_info[pk_range_wrapper]
+            location = global_endpoint_manager.location_cache.get_location_from_endpoint(
+                str(request.location_endpoint_to_route))
+            regional_context = (global_endpoint_manager.location_cache.
+                                account_read_regional_routing_contexts_by_location.get(location).primary_endpoint)
+            partition_level_info.unavailable_regional_endpoints[location] = regional_context
@@ -25,6 +25,7 @@
 
 from typing import Dict
 from typing_extensions import Literal
+# cspell:ignore PPAF
 
 
 class _Constants:
@@ -40,6 +41,7 @@ class _Constants:
     DatabaseAccountEndpoint: Literal["databaseAccountEndpoint"] = "databaseAccountEndpoint"
     DefaultEndpointsRefreshTime: int = 5 * 60 * 1000 # milliseconds
     UnavailableEndpointDBATimeouts: int = 1 # seconds
+    EnablePerPartitionFailoverBehavior: Literal["enablePerPartitionFailoverBehavior"] = "enablePerPartitionFailoverBehavior" #pylint: disable=line-too-long
 
     # ServiceDocument Resource
     EnableMultipleWritableLocations: Literal["enableMultipleWriteLocations"] = "enableMultipleWriteLocations"
@@ -71,6 +73,10 @@ class _Constants:
     FAILURE_PERCENTAGE_TOLERATED = "AZURE_COSMOS_FAILURE_PERCENTAGE_TOLERATED"
     FAILURE_PERCENTAGE_TOLERATED_DEFAULT: int = 90
     # -------------------------------------------------------------------------
+    # Only applicable when per partition automatic failover is enabled --------
+    TIMEOUT_ERROR_THRESHOLD_PPAF = "AZURE_COSMOS_TIMEOUT_ERROR_THRESHOLD_FOR_PPAF"
+    TIMEOUT_ERROR_THRESHOLD_PPAF_DEFAULT: int = 10
+    # -------------------------------------------------------------------------
 
     # Error code translations
     ERROR_TRANSLATIONS: Dict[int, str] = {

@@ -50,7 +50,7 @@
     HttpResponse  # pylint: disable=no-legacy-azure-core-http-response-import
 
 from . import _base as base
-from ._global_partition_endpoint_manager_circuit_breaker import _GlobalPartitionEndpointManagerForCircuitBreaker
+from ._global_partition_endpoint_manager_per_partition_automatic_failover import _GlobalPartitionEndpointManagerForPerPartitionAutomaticFailover # pylint: disable=line-too-long
 from . import _query_iterable as query_iterable
 from . import _runtime_constants as runtime_constants
 from . import _session
@@ -170,7 +170,7 @@ def __init__( # pylint: disable=too-many-statements
         self.last_response_headers: CaseInsensitiveDict = CaseInsensitiveDict()
 
         self.UseMultipleWriteLocations = False
-        self._global_endpoint_manager = _GlobalPartitionEndpointManagerForCircuitBreaker(self)
+        self._global_endpoint_manager = _GlobalPartitionEndpointManagerForPerPartitionAutomaticFailover(self)
 
         retry_policy = None
         if isinstance(self.connection_policy.ConnectionRetryConfiguration, HTTPPolicy):
@@ -2677,12 +2677,15 @@ def GetDatabaseAccount(
             database_account._ReadableLocations = result[Constants.ReadableLocations]
         if Constants.EnableMultipleWritableLocations in result:
             database_account._EnableMultipleWritableLocations = result[
-                Constants.EnableMultipleWritableLocations
-            ]
+                Constants.EnableMultipleWritableLocations]
 
         self.UseMultipleWriteLocations = (
                 self.connection_policy.UseMultipleWriteLocations and database_account._EnableMultipleWritableLocations
         )
+
+        if Constants.EnablePerPartitionFailoverBehavior in result:
+            database_account._EnablePerPartitionFailoverBehavior = result[Constants.EnablePerPartitionFailoverBehavior]
+
         if response_hook:
             response_hook(last_response_headers, result)
         return database_account

@@ -23,17 +23,8 @@
 Azure Cosmos database service.
 """
 
-import logging
 from azure.cosmos.documents import _OperationType
 
-logger = logging.getLogger(__name__)
-logger.setLevel(logging.INFO)
-log_formatter = logging.Formatter("%(levelname)s:%(message)s")
-log_handler = logging.StreamHandler()
-log_handler.setFormatter(log_formatter)
-logger.addHandler(log_handler)
-
-
 class EndpointDiscoveryRetryPolicy(object):
     """The endpoint discovery retry policy class used for geo-replicated database accounts
        to handle the write forbidden exceptions due to writable/readable location changes
@@ -43,8 +34,9 @@ class EndpointDiscoveryRetryPolicy(object):
     Max_retry_attempt_count = 120
     Retry_after_in_milliseconds = 1000
 
-    def __init__(self, connection_policy, global_endpoint_manager, *args):
+    def __init__(self, connection_policy, global_endpoint_manager, pk_range_wrapper, *args):
         self.global_endpoint_manager = global_endpoint_manager
+        self.pk_range_wrapper = pk_range_wrapper
         self._max_retry_attempt_count = EndpointDiscoveryRetryPolicy.Max_retry_attempt_count
         self.failover_retry_count = 0
         self.retry_after_in_milliseconds = EndpointDiscoveryRetryPolicy.Retry_after_in_milliseconds
@@ -70,6 +62,22 @@ def ShouldRetry(self, exception):  # pylint: disable=unused-argument
 
         self.failover_retry_count += 1
 
+        # set the refresh_needed flag to ensure that endpoint list is
+        # refreshed with new writable and readable locations
+        self.global_endpoint_manager.refresh_needed = True
+
+        # If per partition automatic failover is applicable, we mark the current endpoint as unavailable
+        # and resolve the service endpoint for the partition range - otherwise, continue the default retry logic
+        if self.global_endpoint_manager.is_per_partition_automatic_failover_applicable(self.request):
+            partition_level_info = self.global_endpoint_manager.partition_range_to_failover_info[self.pk_range_wrapper]
+            location = self.global_endpoint_manager.location_cache.get_location_from_endpoint(
+                str(self.request.location_endpoint_to_route))
+            location_endpoint = (self.global_endpoint_manager.location_cache.
+                                account_read_regional_routing_contexts_by_location.get(location).primary_endpoint)
+            partition_level_info.unavailable_regional_endpoints[location] = location_endpoint
+            self.global_endpoint_manager.resolve_service_endpoint_for_partition(self.request, self.pk_range_wrapper)
+            return True
+
         if self.request.location_endpoint_to_route:
             if _OperationType.IsReadOnlyOperation(self.request.operation_type):
                 # Mark current read endpoint as unavailable
@@ -81,10 +89,6 @@ def ShouldRetry(self, exception):  # pylint: disable=unused-argument
                     self.request.location_endpoint_to_route,
                     True)
 
-        # set the refresh_needed flag to ensure that endpoint list is
-        # refreshed with new writable and readable locations
-        self.global_endpoint_manager.refresh_needed = True
-
         # clear previous location-based routing directive
         self.request.clear_route_to_location()
 

@@ -36,6 +36,8 @@
 if TYPE_CHECKING:
     from azure.cosmos._cosmos_client_connection import CosmosClientConnection
 
+#cspell:ignore ppcb
+
 class _GlobalPartitionEndpointManagerForCircuitBreaker(_GlobalEndpointManager):
     """
     This internal class implements the logic for partition endpoint management for
@@ -93,16 +95,17 @@ def create_pk_range_wrapper(self, request: RequestObject) -> Optional[PartitionK
 
         return PartitionKeyRangeWrapper(partition_range, container_rid)
 
-    def record_failure(
+    def record_ppcb_failure(
             self,
-            request: RequestObject
-    ) -> None:
+            request: RequestObject,
+            pk_range_wrapper: Optional[PartitionKeyRangeWrapper] = None)-> None:
         if self.is_circuit_breaker_applicable(request):
-            pk_range_wrapper = self.create_pk_range_wrapper(request)
+            if pk_range_wrapper is None:
+                pk_range_wrapper = self.create_pk_range_wrapper(request)
             if pk_range_wrapper:
                 self.global_partition_endpoint_manager_core.record_failure(request, pk_range_wrapper)
 
-    def resolve_service_endpoint_for_partition(
+    def _resolve_service_endpoint_for_partition_circuit_breaker(
             self,
             request: RequestObject,
             pk_range_wrapper: Optional[PartitionKeyRangeWrapper]
@@ -113,11 +116,12 @@ def resolve_service_endpoint_for_partition(
                                                                                                     pk_range_wrapper)
         return self._resolve_service_endpoint(request)
 
-    def record_success(
+    def record_ppcb_success(
             self,
-            request: RequestObject
-    ) -> None:
-        if self.global_partition_endpoint_manager_core.is_circuit_breaker_applicable(request):
-            pk_range_wrapper = self.create_pk_range_wrapper(request)
+            request: RequestObject,
+            pk_range_wrapper: Optional[PartitionKeyRangeWrapper] = None) -> None:
+        if self.is_circuit_breaker_applicable(request):
+            if pk_range_wrapper is None:
+                pk_range_wrapper = self.create_pk_range_wrapper(request)
             if pk_range_wrapper:
                 self.global_partition_endpoint_manager_core.record_success(request, pk_range_wrapper)
@@ -19,6 +19,8 @@
 # OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 # SOFTWARE.
 
+# pylint: disable=protected-access
+
 """Internal class for global endpoint manager for circuit breaker.
 """
 import logging
@@ -60,7 +62,10 @@ def is_circuit_breaker_applicable(self, request: RequestObject) -> bool:
             return False
 
         circuit_breaker_enabled = os.environ.get(Constants.CIRCUIT_BREAKER_ENABLED_CONFIG,
-                                                 Constants.CIRCUIT_BREAKER_ENABLED_CONFIG_DEFAULT) == "True"
+                                                 Constants.CIRCUIT_BREAKER_ENABLED_CONFIG_DEFAULT).lower() == "true"
+        if not circuit_breaker_enabled and self.client._global_endpoint_manager is not None:
+            if self.client._global_endpoint_manager._database_account_cache is not None:
+                circuit_breaker_enabled = self.client._global_endpoint_manager._database_account_cache._EnablePerPartitionFailoverBehavior is True # pylint: disable=line-too-long
         if not circuit_breaker_enabled:
             return False