Pyright 1.1.385 (patroni#3182)

Declaring variables with `Union` and using `isinstance()` hack doesn't work anymore. Therefore the code is updated to use `Any` for variable and `cast` function after firguring out the correct type in order to avoid getting errors about `Unknown` types.

Author: CyberDem0n

.github/workflows/tests.yaml

@@ -186,7 +186,7 @@ jobs:
 
     - uses: jakebailey/pyright-action@v2
       with:
-        version: 1.1.379
+        version: 1.1.385
 
   docs:
     runs-on: ubuntu-latest

patroni/api.py

@@ -21,7 +21,7 @@
 from ipaddress import ip_address, ip_network, IPv4Network, IPv6Network
 from socketserver import ThreadingMixIn
 from threading import Thread
-from typing import Any, Callable, Dict, Iterator, List, Optional, Tuple, TYPE_CHECKING, Union
+from typing import Any, Callable, cast, Dict, Iterator, List, Optional, Tuple, TYPE_CHECKING, Union
 from urllib.parse import parse_qs, urlparse
 
 import dateutil.parser
@@ -71,8 +71,8 @@ def check_access(*args: Any, **kwargs: Any) -> Callable[..., Any]:
     """
     allowlist_check_members = kwargs.get('allowlist_check_members', True)
 
-    def inner_decorator(func: Callable[..., None]) -> Callable[..., None]:
-        def wrapper(self: 'RestApiHandler', *args: Any, **kwargs: Any) -> None:
+    def inner_decorator(func: Callable[..., Any]) -> Callable[..., Any]:
+        def wrapper(self: 'RestApiHandler', *args: Any, **kwargs: Any) -> Any:
             if self.server.check_access(self, allowlist_check_members=allowlist_check_members):
                 return func(self, *args, **kwargs)
 
@@ -698,9 +698,9 @@ def _read_json_content(self, body_is_optional: bool = False) -> Optional[Dict[An
             content_length = int(self.headers.get('content-length') or 0)
             if content_length == 0 and body_is_optional:
                 return {}
-            request: Union[Dict[str, Any], Any] = json.loads(self.rfile.read(content_length).decode('utf-8'))
+            request = json.loads(self.rfile.read(content_length).decode('utf-8'))
             if isinstance(request, dict) and (request or body_is_optional):
-                return request
+                return cast(Dict[str, Any], request)
         except Exception:
             logger.exception('Bad request')
         self.send_error(400)
@@ -1723,12 +1723,11 @@ def get_certificate_serial_number(self) -> Optional[str]:
 
         :returns: serial number of the certificate configured through ``restapi.certfile`` setting.
         """
-        if self.__ssl_options.get('certfile'):
+        certfile: Optional[str] = self.__ssl_options.get('certfile')
+        if certfile:
             import ssl
             try:
-                crt: Dict[str, Any] = ssl._ssl._test_decode_cert(self.__ssl_options['certfile'])  # pyright: ignore
-                if TYPE_CHECKING:  # pragma: no cover
-                    assert isinstance(crt, dict)
+                crt = cast(Dict[str, Any], ssl._ssl._test_decode_cert(certfile))  # pyright: ignore
                 return crt.get('serialNumber')
             except ssl.SSLError as e:
                 logger.error('Failed to get serial number from certificate %s: %r', self.__ssl_options['certfile'], e)

patroni/config.py

@@ -8,7 +8,7 @@
 
 from collections import defaultdict
 from copy import deepcopy
-from typing import Any, Callable, Collection, Dict, List, Optional, TYPE_CHECKING, Union
+from typing import Any, Callable, cast, Collection, Dict, List, Optional, TYPE_CHECKING, Union
 
 import yaml
 
@@ -695,8 +695,8 @@ def _build_effective_configuration(self, dynamic_configuration: Dict[str, Any],
         config = self._safe_copy_dynamic_configuration(dynamic_configuration)
         for name, value in local_configuration.items():
             if name == 'citus':  # remove invalid citus configuration
-                if isinstance(value, dict) and isinstance(value.get('group'), int)\
-                        and isinstance(value.get('database'), str):
+                if isinstance(value, dict) and isinstance(cast(Dict[str, Any], value).get('group'), int) \
+                        and isinstance(cast(Dict[str, Any], value).get('database'), str):
                     config[name] = value
             elif name == 'postgresql':
                 for name, value in (value or {}).items():

patroni/dcs/__init__.py

@@ -10,8 +10,8 @@
 from copy import deepcopy
 from random import randint
 from threading import Event, Lock
-from typing import Any, Callable, Collection, Dict, Iterator, List, \
-    NamedTuple, Optional, Set, Tuple, Type, TYPE_CHECKING, Union
+from typing import Any, Callable, cast, Collection, Dict, Iterator, \
+    List, NamedTuple, Optional, Set, Tuple, Type, TYPE_CHECKING, Union
 from urllib.parse import parse_qsl, urlparse, urlunparse
 
 import dateutil.parser
@@ -219,7 +219,7 @@ def conn_url(self) -> Optional[str]:
 
         return None
 
-    def conn_kwargs(self, auth: Union[Any, Dict[str, Any], None] = None) -> Dict[str, Any]:
+    def conn_kwargs(self, auth: Optional[Any] = None) -> Dict[str, Any]:
         """Give keyword arguments used for PostgreSQL connection settings.
 
         :param auth: Authentication properties - can be defined as anything supported by the ``psycopg2`` or
@@ -255,7 +255,7 @@ def conn_kwargs(self, auth: Union[Any, Dict[str, Any], None] = None) -> Dict[str
 
         # apply any remaining authentication parameters
         if auth and isinstance(auth, dict):
-            ret.update({k: v for k, v in auth.items() if v is not None})
+            ret.update({k: v for k, v in cast(Dict[str, Any], auth).items() if v is not None})
             if 'username' in auth:
                 ret['user'] = ret.pop('username')
         return ret
@@ -949,28 +949,28 @@ def get_clone_member(self, exclude_name: str) -> Union[Member, Leader, None]:
         return candidates[randint(0, len(candidates) - 1)] if candidates else self.leader
 
     @staticmethod
-    def is_physical_slot(value: Union[Any, Dict[str, Any]]) -> bool:
+    def is_physical_slot(value: Any) -> bool:
         """Check whether provided configuration is for permanent physical replication slot.
 
         :param value: configuration of the permanent replication slot.
 
         :returns: ``True`` if *value* is a physical replication slot, otherwise ``False``.
         """
         return not value \
-            or (isinstance(value, dict) and not Cluster.is_logical_slot(value)
-                and value.get('type', 'physical') == 'physical')
+            or (isinstance(value, dict) and not Cluster.is_logical_slot(cast(Dict[str, Any], value))
+                and cast(Dict[str, Any], value).get('type', 'physical') == 'physical')
 
     @staticmethod
-    def is_logical_slot(value: Union[Any, Dict[str, Any]]) -> bool:
+    def is_logical_slot(value: Any) -> bool:
         """Check whether provided configuration is for permanent logical replication slot.
 
         :param value: configuration of the permanent replication slot.
 
         :returns: ``True`` if *value* is a logical replication slot, otherwise ``False``.
         """
         return isinstance(value, dict) \
-            and value.get('type', 'logical') == 'logical' \
-            and bool(value.get('database') and value.get('plugin'))
+            and cast(Dict[str, Any], value).get('type', 'logical') == 'logical' \
+            and bool(cast(Dict[str, Any], value).get('database') and cast(Dict[str, Any], value).get('plugin'))
 
     @property
     def __permanent_slots(self) -> Dict[str, Union[Dict[str, Any], Any]]:
@@ -992,7 +992,7 @@ def __permanent_slots(self) -> Dict[str, Union[Dict[str, Any], Any]]:
                     value['lsn'] = lsn
                 else:
                     # Don't let anyone set 'lsn' in the global configuration :)
-                    value.pop('lsn', None)
+                    value.pop('lsn', None)  # pyright: ignore [reportUnknownMemberType]
         return ret
 
     @property
@@ -1066,8 +1066,9 @@ def _merge_permanent_slots(self, slots: Dict[str, Dict[str, Any]], permanent_slo
                 logger.error("Slot name may only contain lower case letters, numbers, and the underscore chars")
                 continue
 
-            value = deepcopy(value) if value else {'type': 'physical'}
-            if isinstance(value, dict):
+            tmp = deepcopy(value) if value else {'type': 'physical'}
+            if isinstance(tmp, dict):
+                value = cast(Dict[str, Any], tmp)
                 if 'type' not in value:
                     value['type'] = 'logical' if value.get('database') and value.get('plugin') else 'physical'
 

patroni/dcs/etcd3.py

@@ -150,8 +150,7 @@ class InvalidAuthToken(Etcd3ClientError):
 errCodeToClientError = {getattr(s, 'code'): s for s in Etcd3ClientError.__subclasses__()}
 
 
-def _raise_for_data(data: Union[bytes, str, Dict[str, Union[Any, Dict[str, Any]]]],
-                    status_code: Optional[int] = None) -> Etcd3ClientError:
+def _raise_for_data(data: Union[bytes, str, Dict[str, Any]], status_code: Optional[int] = None) -> Etcd3ClientError:
     try:
         if TYPE_CHECKING:  # pragma: no cover
             assert isinstance(data, dict)

patroni/dcs/exhibitor.py

@@ -3,7 +3,7 @@
 import random
 import time
 
-from typing import Any, Callable, Dict, List, Union
+from typing import Any, Callable, cast, Dict, List, Union
 
 from ..postgresql.mpp import AbstractMPP
 from ..request import get as requests_get
@@ -41,16 +41,17 @@ def poll(self) -> bool:
 
         if isinstance(json, dict) and 'servers' in json and 'port' in json:
             self._next_poll = time.time() + self._poll_interval
-            servers: List[str] = json['servers']
-            zookeeper_hosts = ','.join([h + ':' + str(json['port']) for h in sorted(servers)])
+            servers: List[str] = cast(Dict[str, Any], json)['servers']
+            port = str(cast(Dict[str, Any], json)['port'])
+            zookeeper_hosts = ','.join([h + ':' + port for h in sorted(servers)])
             if self._zookeeper_hosts != zookeeper_hosts:
                 logger.info('ZooKeeper connection string has changed: %s => %s', self._zookeeper_hosts, zookeeper_hosts)
                 self._zookeeper_hosts = zookeeper_hosts
                 self._exhibitors = json['servers']
                 return True
         return False
 
-    def _query_exhibitors(self, exhibitors: List[str]) -> Union[Dict[str, Any], Any]:
+    def _query_exhibitors(self, exhibitors: List[str]) -> Any:
         random.shuffle(exhibitors)
         for host in exhibitors:
             try:

patroni/dcs/kubernetes.py

@@ -648,7 +648,7 @@ def get(self, name: str) -> Optional[K8sObject]:
         with self._object_cache_lock:
             return self._object_cache.get(name)
 
-    def _process_event(self, event: Dict[str, Union[Any, Dict[str, Union[Any, Dict[str, Any]]]]]) -> None:
+    def _process_event(self, event: Dict[str, Any]) -> None:
         ev_type = event['type']
         obj = event['object']
         name = obj['metadata']['name']

patroni/dcs/raft.py

@@ -255,7 +255,7 @@ def callback(*args: Any) -> None:
                         self.__limb.pop(key)
                 self._expire(key, value, callback=callback)
 
-    def get(self, key: str, recursive: bool = False) -> Union[None, Dict[str, Any], Dict[str, Dict[str, Any]]]:
+    def get(self, key: str, recursive: bool = False) -> Optional[Dict[str, Any]]:
         if not recursive:
             return self.__data.get(key)
         return {k: v for k, v in self.__data.items() if k.startswith(key)}

patroni/dcs/zookeeper.py

@@ -4,7 +4,7 @@
 import socket
 import time
 
-from typing import Any, Callable, Dict, List, Optional, Tuple, TYPE_CHECKING, Union
+from typing import Any, Callable, cast, Dict, List, Optional, Tuple, TYPE_CHECKING, Union
 
 from kazoo.client import KazooClient, KazooRetry, KazooState
 from kazoo.exceptions import ConnectionClosedError, NodeExistsError, NoNodeError, SessionExpiredError
@@ -180,7 +180,8 @@ def ttl(self) -> int:
         return int(self._client._session_timeout / 1000.0)
 
     def set_retry_timeout(self, retry_timeout: int) -> None:
-        retry = self._client.retry if isinstance(self._client.retry, KazooRetry) else self._client._retry
+        old_kazoo = isinstance(self._client.retry, KazooRetry)  # pyright: ignore [reportUnnecessaryIsInstance]
+        retry = cast(KazooRetry, self._client.retry) if old_kazoo else self._client._retry
         retry.deadline = retry_timeout
 
     def get_node(

patroni/global_config.py

@@ -8,7 +8,7 @@
 import types
 
 from copy import deepcopy
-from typing import Any, Dict, List, Optional, TYPE_CHECKING, Union
+from typing import Any, cast, Dict, List, Optional, TYPE_CHECKING
 
 from .collections import EMPTY_DICT
 from .utils import parse_bool, parse_int
@@ -121,7 +121,7 @@ def is_synchronous_mode_strict(self) -> bool:
         """``True`` if at least one synchronous node is required."""
         return self.check_mode('synchronous_mode_strict')
 
-    def get_standby_cluster_config(self) -> Union[Dict[str, Any], Any]:
+    def get_standby_cluster_config(self) -> Any:
         """Get ``standby_cluster`` configuration.
 
         :returns: a copy of ``standby_cluster`` configuration.
@@ -133,7 +133,7 @@ def is_standby_cluster(self) -> bool:
         """``True`` if global configuration has a valid ``standby_cluster`` section."""
         config = self.get_standby_cluster_config()
         return isinstance(config, dict) and\
-            bool(config.get('host') or config.get('port') or config.get('restore_command'))
+            any(cast(Dict[str, Any], config).get(p) for p in ('host', 'port', 'restore_command'))
 
     def get_int(self, name: str, default: int = 0, base_unit: Optional[str] = None) -> int:
         """Gets current value of *name* from the global configuration and try to return it as :class:`int`.