diff --git a/pymongo/max_staleness_selectors.py b/pymongo/max_staleness_selectors.py index 28b0bb615..2b7a7cc3c 100644 --- a/pymongo/max_staleness_selectors.py +++ b/pymongo/max_staleness_selectors.py @@ -26,17 +26,22 @@ When there is no known primary, a secondary S's staleness is estimated with: where "SMax" is the secondary with the greatest lastWriteDate. """ +from __future__ import annotations + +from typing import TYPE_CHECKING from pymongo.errors import ConfigurationError from pymongo.server_type import SERVER_TYPE +if TYPE_CHECKING: + from pymongo.server_selectors import Selection # Constant defined in Max Staleness Spec: An idle primary writes a no-op every # 10 seconds to refresh secondaries' lastWriteDate values. IDLE_WRITE_PERIOD = 10 SMALLEST_MAX_STALENESS = 90 -def _validate_max_staleness(max_staleness, heartbeat_frequency): +def _validate_max_staleness(max_staleness: int, heartbeat_frequency: int) -> None: # We checked for max staleness -1 before this, it must be positive here. if max_staleness < heartbeat_frequency + IDLE_WRITE_PERIOD: raise ConfigurationError( @@ -53,7 +58,7 @@ def _validate_max_staleness(max_staleness, heartbeat_frequency): ) -def _with_primary(max_staleness, selection): +def _with_primary(max_staleness: int, selection: Selection) -> Selection: """Apply max_staleness, in seconds, to a Selection with a known primary.""" primary = selection.primary sds = [] @@ -75,7 +80,7 @@ def _with_primary(max_staleness, selection): return selection.with_server_descriptions(sds) -def _no_primary(max_staleness, selection): +def _no_primary(max_staleness: int, selection: Selection) -> Selection: """Apply max_staleness, in seconds, to a Selection with no known primary.""" # Secondary that's replicated the most recent writes. smax = selection.secondary_with_max_last_write_date() @@ -98,7 +103,7 @@ def _no_primary(max_staleness, selection): return selection.with_server_descriptions(sds) -def select(max_staleness, selection): +def select(max_staleness: int, selection: Selection) -> Selection: """Apply max_staleness, in seconds, to a Selection.""" if max_staleness == -1: return selection