PYTHON-3787 add types to max_staleness_selectors.py (#1276)

This commit is contained in:
Iris 2023-06-28 09:48:57 -07:00 committed by GitHub
parent 7d19205540
commit ebba342aaa
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23

View File

@ -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