Add a SyncType in type.py to avoid repeating the synctype declaration

Author: blogh

check_patroni/cli.py

@@ -1,7 +1,7 @@
 import logging
 import re
 from configparser import ConfigParser
-from typing import List, Literal
+from typing import List
 
 import click
 import nagiosplugin
@@ -34,7 +34,7 @@
     NodeTLHasChanged,
     NodeTLHasChangedSummary,
 )
-from .types import ConnectionInfo, Parameters
+from .types import ConnectionInfo, Parameters, SyncType
 
 DEFAULT_CFG = "config.ini"
 handler = logging.StreamHandler()
@@ -377,7 +377,7 @@ def cluster_has_replica(
     critical: str,
     sync_warning: str,
     sync_critical: str,
-    sync_type: Literal["any", "sync", "quorum"],
+    sync_type: SyncType,
     max_lag: str,
 ) -> None:
     """Check if the cluster has healthy replicas and/or if some are sync or quorum standbies
@@ -647,7 +647,7 @@ def node_is_replica(
     max_lag: str,
     check_is_sync: bool,
     check_is_async: bool,
-    sync_type: Literal["any", "sync", "quorum"],
+    sync_type: SyncType,
 ) -> None:
     """Check if the node is a replica with no noloadbalance tag.
 

check_patroni/cluster.py

@@ -1,12 +1,12 @@
 import hashlib
 import json
 from collections import Counter
-from typing import Any, Iterable, Literal, Union
+from typing import Any, Iterable, Union
 
 import nagiosplugin
 
 from . import _log
-from .types import ConnectionInfo, PatroniResource, handle_unknown
+from .types import ConnectionInfo, PatroniResource, SyncType, handle_unknown
 
 
 def replace_chars(text: str) -> str:
@@ -137,7 +137,7 @@ def __init__(
         self,
         connection_info: ConnectionInfo,
         max_lag: Union[int, None],
-        sync_type: Literal["any", "sync", "quorum"],
+        sync_type: SyncType,
     ):
         super().__init__(connection_info)
         self.max_lag = max_lag

check_patroni/node.py

@@ -1,9 +1,9 @@
-from typing import Iterable, Literal
+from typing import Iterable
 
 import nagiosplugin
 
 from . import _log
-from .types import APIError, ConnectionInfo, PatroniResource, handle_unknown
+from .types import APIError, ConnectionInfo, PatroniResource, SyncType, handle_unknown
 
 
 class NodeIsPrimary(PatroniResource):
@@ -65,7 +65,7 @@ def __init__(
         max_lag: str,
         check_is_sync: bool,
         check_is_async: bool,
-        sync_type: Literal["any", "sync", "quorum"],
+        sync_type: SyncType,
     ) -> None:
         super().__init__(connection_info)
         self.max_lag = max_lag
@@ -105,7 +105,7 @@ def __init__(
         lag: str,
         check_is_sync: bool,
         check_is_async: bool,
-        sync_type: Literal["any", "sync", "quorum"],
+        sync_type: SyncType,
     ) -> None:
         self.lag = lag
         if check_is_sync:

check_patroni/types.py

@@ -1,6 +1,6 @@
 import json
 from functools import lru_cache
-from typing import Any, Callable, List, Optional, Tuple, Union
+from typing import Any, Callable, List, Literal, Optional, Tuple, Union
 from urllib.parse import urlparse
 
 import attr
@@ -9,6 +9,8 @@
 
 from . import _log
 
+SyncType = Literal["any", "sync", "quorum"]
+
 
 class APIError(requests.exceptions.RequestException):
     """This exception is raised when the rest api could