Pyright 1.1.385 (#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.

.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 http.server import BaseHTTPRequestHandler, HTTPServer
 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 @@ class RestApiHandler(BaseHTTPRequestHandler):
             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 @@ class RestApiServer(ThreadingMixIn, HTTPServer, Thread):
 
         :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 @@ import tempfile
 
 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 @@ class Config(object):
         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 collections import defaultdict
 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 @@ class Member(Tags, NamedTuple('Member',
 
         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 @@ class Member(Tags, NamedTuple('Member',
 
         # 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,7 +949,7 @@ class Cluster(NamedTuple('Cluster',
         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.
@@ -957,11 +957,11 @@ class Cluster(NamedTuple('Cluster',
         :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.
@@ -969,8 +969,8 @@ class Cluster(NamedTuple('Cluster',
         :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 @@ class Cluster(NamedTuple('Cluster',
                     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 @@ class Cluster(NamedTuple('Cluster',
                 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 @@ errStringToClientError = {getattr(s, 'error'): s for s in Etcd3ClientError.get_s
 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 logging
 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,8 +41,9 @@ class ExhibitorEnsembleProvider(object):
 
         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
@@ -50,7 +51,7 @@ class ExhibitorEnsembleProvider(object):
                 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 @@ class ObjectCache(Thread):
         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 @@ class KVStoreTTL(DynMemberSyncObj):
                         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 select
 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 @@ class ZooKeeper(AbstractDCS):
         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 sys
 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 @@ class GlobalConfig(types.ModuleType):
         """``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 @@ class GlobalConfig(types.ModuleType):
         """``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`.

patroni/log.py

@@ -12,7 +12,7 @@ from io import TextIOWrapper
 from logging.handlers import RotatingFileHandler
 from queue import Full, Queue
 from threading import Lock, Thread
-from typing import Any, Dict, List, Optional, TYPE_CHECKING, Union
+from typing import Any, cast, Dict, List, Optional, TYPE_CHECKING, Union
 
 from .file_perm import pg_perm
 from .utils import deep_compare, parse_int
@@ -358,6 +358,7 @@ class PatroniLogger(Thread):
             jsonformat = logformat
             rename_fields = {}
         elif isinstance(logformat, list):
+            logformat = cast(List[Any], logformat)
             log_fields: List[str] = []
             rename_fields: Dict[str, str] = {}
 
@@ -365,6 +366,7 @@ class PatroniLogger(Thread):
                 if isinstance(field, str):
                     log_fields.append(field)
                 elif isinstance(field, dict):
+                    field = cast(Dict[str, Any], field)
                     for original_field, renamed_field in field.items():
                         if isinstance(renamed_field, str):
                             log_fields.append(original_field)

patroni/postgresql/bootstrap.py

@@ -4,7 +4,7 @@ import shlex
 import tempfile
 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 ..async_executor import CriticalTask
 from ..collections import EMPTY_DICT
@@ -33,8 +33,7 @@ class Bootstrap(object):
         return self._running_custom_bootstrap and self._keep_existing_recovery_conf
 
     @staticmethod
-    def process_user_options(tool: str,
-                             options: Union[Any, Dict[str, str], List[Union[str, Dict[str, Any]]]],
+    def process_user_options(tool: str, options: Any,
                              not_allowed_options: Tuple[str, ...],
                              error_handler: Callable[[str], None]) -> List[str]:
         """Format *options* in a list or dictionary format into command line long form arguments.
@@ -92,20 +91,21 @@ class Bootstrap(object):
             return ret
 
         if isinstance(options, dict):
-            for key, val in options.items():
+            for key, val in cast(Dict[str, str], options).items():
                 if key and val:
                     user_options.append('--{0}={1}'.format(key, unquote(val)))
         elif isinstance(options, list):
-            for opt in options:
+            for opt in cast(List[Any], options):
                 if isinstance(opt, str) and option_is_allowed(opt):
                     user_options.append('--{0}'.format(opt))
                 elif isinstance(opt, dict):
-                    keys = list(opt.keys())
-                    if len(keys) == 1 and isinstance(opt[keys[0]], str) and option_is_allowed(keys[0]):
-                        user_options.append('--{0}={1}'.format(keys[0], unquote(opt[keys[0]])))
+                    args = cast(Dict[str, Any], opt)
+                    keys = list(args.keys())
+                    if len(keys) == 1 and isinstance(args[keys[0]], str) and option_is_allowed(keys[0]):
+                        user_options.append('--{0}={1}'.format(keys[0], unquote(args[keys[0]])))
                     else:
                         error_handler('Error when parsing {0} key-value option {1}: only one key-value is allowed'
-                                      ' and value should be a string'.format(tool, opt[keys[0]]))
+                                      ' and value should be a string'.format(tool, args[keys[0]]))
                 else:
                     error_handler('Error when parsing {0} option {1}: value should be string value'
                                   ' or a single key-value pair'.format(tool, opt))

patroni/postgresql/cancellable.py

@@ -2,7 +2,7 @@ import logging
 import subprocess
 
 from threading import Lock
-from typing import Any, Dict, List, Optional, Union
+from typing import Any, Dict, List, Optional
 
 import psutil
 
@@ -76,7 +76,7 @@ class CancellableSubprocess(CancellableExecutor):
         super(CancellableSubprocess, self).__init__()
         self._is_cancelled = False
 
-    def call(self, *args: Any, **kwargs: Union[Any, Dict[str, str]]) -> Optional[int]:
+    def call(self, *args: Any, **kwargs: Any) -> Optional[int]:
         for s in ('stdin', 'stdout', 'stderr'):
             kwargs.pop(s, None)
 

patroni/postgresql/mpp/citus.py

@@ -3,7 +3,7 @@ import re
 import time
 
 from threading import Condition, Event, Thread
-from typing import Any, Collection, Dict, Iterator, List, Optional, Set, Tuple, TYPE_CHECKING, Union
+from typing import Any, cast, Collection, Dict, Iterator, List, Optional, Set, Tuple, TYPE_CHECKING, Union
 from urllib.parse import urlparse
 
 from ...dcs import Cluster
@@ -359,7 +359,7 @@ class Citus(AbstractMPP):
     group_re = re.compile('^(0|[1-9][0-9]*)$')
 
     @staticmethod
-    def validate_config(config: Union[Any, Dict[str, Union[str, int]]]) -> bool:
+    def validate_config(config: Any) -> bool:
         """Check whether provided config is good for a given MPP.
 
         :param config: configuration of ``citus`` MPP section.
@@ -367,8 +367,8 @@ class Citus(AbstractMPP):
         :returns: ``True`` is config passes validation, otherwise ``False``.
         """
         return isinstance(config, dict) \
-            and isinstance(config.get('database'), str) \
-            and parse_int(config.get('group')) is not None
+            and isinstance(cast(Dict[str, Any], config).get('database'), str) \
+            and parse_int(cast(Dict[str, Any], config).get('group')) is not None
 
     @property
     def group(self) -> int:

patroni/utils.py

@@ -25,7 +25,7 @@ import time
 from collections import OrderedDict
 from json import JSONDecoder
 from shlex import split
-from typing import Any, Callable, Dict, Iterator, List, Optional, Tuple, Type, TYPE_CHECKING, Union
+from typing import Any, Callable, cast, Dict, Iterator, List, Optional, Tuple, Type, TYPE_CHECKING, Union
 
 from dateutil import tz
 from urllib3.response import HTTPResponse
@@ -79,7 +79,7 @@ def get_conversion_table(base_unit: str) -> Dict[str, Dict[str, Union[int, float
     return OrderedDict()
 
 
-def deep_compare(obj1: Dict[Any, Union[Any, Dict[Any, Any]]], obj2: Dict[Any, Union[Any, Dict[Any, Any]]]) -> bool:
+def deep_compare(obj1: Dict[Any, Any], obj2: Dict[Any, Any]) -> bool:
     """Recursively compare two dictionaries to check if they are equal in terms of keys and values.
 
     .. note::
@@ -112,14 +112,14 @@ def deep_compare(obj1: Dict[Any, Union[Any, Dict[Any, Any]]], obj2: Dict[Any, Un
 
     for key, value in obj1.items():
         if isinstance(value, dict):
-            if not (isinstance(obj2[key], dict) and deep_compare(value, obj2[key])):
+            if not (isinstance(obj2[key], dict) and deep_compare(cast(Dict[Any, Any], value), obj2[key])):
                 return False
         elif str(value) != str(obj2[key]):
             return False
     return True
 
 
-def patch_config(config: Dict[Any, Union[Any, Dict[Any, Any]]], data: Dict[Any, Union[Any, Dict[Any, Any]]]) -> bool:
+def patch_config(config: Dict[Any, Any], data: Dict[Any, Any]) -> bool:
     """Update and append to dictionary *config* from overrides in *data*.
 
     .. note::
@@ -142,7 +142,7 @@ def patch_config(config: Dict[Any, Union[Any, Dict[Any, Any]]], data: Dict[Any,
         elif name in config:
             if isinstance(value, dict):
                 if isinstance(config[name], dict):
-                    if patch_config(config[name], value):
+                    if patch_config(config[name], cast(Dict[Any, Any], value)):
                         is_changed = True
                 else:
                     config[name] = value
@@ -156,7 +156,7 @@ def patch_config(config: Dict[Any, Union[Any, Dict[Any, Any]]], data: Dict[Any,
     return is_changed
 
 
-def parse_bool(value: Any) -> Union[bool, None]:
+def parse_bool(value: Any) -> Optional[bool]:
     """Parse a given value to a :class:`bool` object.
 
     .. note::
@@ -186,7 +186,7 @@ def parse_bool(value: Any) -> Union[bool, None]:
         return False
 
 
-def strtol(value: Any, strict: Optional[bool] = True) -> Tuple[Union[int, None], str]:
+def strtol(value: Any, strict: Optional[bool] = True) -> Tuple[Optional[int], str]:
     """Extract the long integer part from the beginning of a string that represents a configuration value.
 
     As most as possible close equivalent of ``strtol(3)`` C function (with base=0), which is used by postgres to parse
@@ -240,7 +240,7 @@ def strtol(value: Any, strict: Optional[bool] = True) -> Tuple[Union[int, None],
     return (None if strict else 1), value
 
 
-def strtod(value: Any) -> Tuple[Union[float, None], str]:
+def strtod(value: Any) -> Tuple[Optional[float], str]:
     """Extract the double precision part from the beginning of a string that reprensents a configuration value.
 
     As most as possible close equivalent of ``strtod(3)`` C function, which is used by postgres to parse parameter

patroni/validator.py

@@ -9,7 +9,7 @@ import os
 import shutil
 import socket
 
-from typing import Any, Dict, Iterator, List, Optional as OptionalType, Tuple, TYPE_CHECKING, Union
+from typing import Any, cast, Dict, Iterator, List, Optional as OptionalType, Tuple, TYPE_CHECKING, Union
 
 from .collections import CaseInsensitiveSet, EMPTY_DICT
 from .dcs import dcs_modules
@@ -29,7 +29,7 @@ def populate_validate_params(ignore_listen_port: bool = False) -> None:
     _validation_params['ignore_listen_port'] = ignore_listen_port
 
 
-def validate_log_field(field: Union[str, Dict[str, Any], Any]) -> bool:
+def validate_log_field(field: Any) -> bool:
     """Checks if log field is valid.
 
     :param field: A log field to be validated.
@@ -40,6 +40,7 @@ def validate_log_field(field: Union[str, Dict[str, Any], Any]) -> bool:
     if isinstance(field, str):
         return True
     elif isinstance(field, dict):
+        field = cast(Dict[str, Any], field)
         return len(field) == 1 and isinstance(next(iter(field.values())), str)
     return False
 
@@ -61,6 +62,7 @@ def validate_log_format(logformat: type_logformat) -> bool:
     if isinstance(logformat, str):
         return True
     elif isinstance(logformat, list):
+        logformat = cast(List[Any], logformat)
         if len(logformat) == 0:
             raise ConfigParseError('should contain at least one item')
         if not all(map(validate_log_field, logformat)):
@@ -254,9 +256,8 @@ def get_bin_name(bin_name: str) -> str:
 
     :returns: value of ``postgresql.bin_name[*bin_name*]``, if present, otherwise *bin_name*.
     """
-    if TYPE_CHECKING:  # pragma: no cover
-        assert isinstance(schema.data, dict)
-    return (schema.data.get('postgresql', {}).get('bin_name', {}) or EMPTY_DICT).get(bin_name, bin_name)
+    data = cast(Dict[Any, Any], schema.data)
+    return (data.get('postgresql', {}).get('bin_name', {}) or EMPTY_DICT).get(bin_name, bin_name)
 
 
 def validate_data_dir(data_dir: str) -> bool:
@@ -295,9 +296,8 @@ def validate_data_dir(data_dir: str) -> bool:
             if not os.path.isdir(os.path.join(data_dir, waldir)):
                 raise ConfigParseError("data dir for the cluster is not empty, but doesn't contain"
                                        " \"{}\" directory".format(waldir))
-            if TYPE_CHECKING:  # pragma: no cover
-                assert isinstance(schema.data, dict)
-            bin_dir = schema.data.get("postgresql", {}).get("bin_dir", None)
+            data = cast(Dict[Any, Any], schema.data)
+            bin_dir = data.get("postgresql", {}).get("bin_dir", None)
             major_version = get_major_version(bin_dir, get_bin_name('postgres'))
             if pgversion != major_version:
                 raise ConfigParseError("data_dir directory postgresql version ({}) doesn't match with "
@@ -332,9 +332,8 @@ def validate_binary_name(bin_name: str) -> bool:
     """
     if not bin_name:
         raise ConfigParseError("is an empty string")
-    if TYPE_CHECKING:  # pragma: no cover
-        assert isinstance(schema.data, dict)
-    bin_dir = schema.data.get('postgresql', {}).get('bin_dir', None)
+    data = cast(Dict[Any, Any], schema.data)
+    bin_dir = data.get('postgresql', {}).get('bin_dir', None)
     if not shutil.which(bin_name, path=bin_dir):
         raise ConfigParseError(f"does not contain '{bin_name}' in '{bin_dir or '$PATH'}'")
     return True
@@ -674,7 +673,7 @@ class Schema(object):
                 errors.append(str(i))
         return errors
 
-    def validate(self, data: Union[Dict[Any, Any], Any]) -> Iterator[Result]:
+    def validate(self, data: Any) -> Iterator[Result]:
         """Perform all validations from the schema against the given configuration.
 
         It first checks that *data* argument type is compliant with the type of ``validator`` attribute.
@@ -703,9 +702,9 @@ class Schema(object):
                          "is not {}".format(_get_type_name(self.validator)), level=1, data=self.data)
         elif callable(self.validator):
             if hasattr(self.validator, "expected_type"):
-                if not isinstance(data, self.validator.expected_type):
-                    yield Result(False, "is not {}"
-                                 .format(_get_type_name(self.validator.expected_type)), level=1, data=self.data)
+                expected_type = getattr(self.validator, 'expected_type')
+                if not isinstance(data, expected_type):
+                    yield Result(False, "is not {}".format(_get_type_name(expected_type)), level=1, data=self.data)
                     return
             try:
                 self.validator(data)
@@ -715,41 +714,38 @@ class Schema(object):
         elif isinstance(self.validator, dict):
             if not isinstance(self.data, dict):
                 yield Result(isinstance(self.data, dict), "is not a dictionary", level=1, data=self.data)
-        elif isinstance(self.validator, list):
-            if not isinstance(self.data, list):
-                yield Result(isinstance(self.data, list), "is not a list", level=1, data=self.data)
-                return
-        yield from self.iter()
-
-    def iter(self) -> Iterator[Result]:
-        """Iterate over ``validator``, if it is an iterable object, to validate the corresponding entries in ``data``.
-
-        Only :class:`dict`, :class:`list`, :class:`Directory` and :class:`Or` objects are considered iterable objects.
-
-        :yields: objects with the error message related to the failure, if any check fails.
-        """
-        if isinstance(self.validator, dict):
-            if not isinstance(self.data, dict):
-                yield Result(False, "is not a dictionary.", level=1)
             else:
                 yield from self.iter_dict()
         elif isinstance(self.validator, list):
-            if len(self.data) == 0:
-                yield Result(False, "is an empty list", data=self.data)
-            if self.validator:
-                for key, value in enumerate(self.data):
-                    # Although the value in the configuration (`data`) is expected to contain 1 or more entries, only
-                    # the first validator defined in `validator` property list will be used. It is only defined as a
-                    # `list` in `validator` so this logic can understand that the value in `data` attribute should be a
-                    # `list`. For example: "pg_hba": [str] in `validator` attribute defines that "pg_hba" in `data`
-                    # attribute should contain a list with one or more `str` entries.
-                    for v in Schema(self.validator[0]).validate(value):
-                        yield Result(v.status, v.error,
-                                     path=(str(key) + ("." + v.path if v.path else "")), level=v.level, data=value)
-        elif isinstance(self.validator, Directory) and isinstance(self.data, str):
-            yield from self.validator.validate(self.data)
+            if not isinstance(self.data, list):
+                yield Result(isinstance(self.data, list), "is not a list", level=1, data=self.data)
+            else:
+                yield from self.iter_list()
         elif isinstance(self.validator, Or):
             yield from self.iter_or()
+        elif isinstance(self.validator, Directory) and isinstance(self.data, str):
+            yield from self.validator.validate(self.data)
+
+    def iter_list(self) -> Iterator[Result]:
+        """Iterate over a ``data`` object and perform validations using the first element of the ``validator``.
+
+        :yields: objects with the error message related to the failure, if any check fails.
+        """
+        data = cast(List[Any], self.data)
+        if len(data) == 0:
+            yield Result(False, "is an empty list", data=data)
+
+        validators = cast(List[Any], self.validator)
+        if len(validators):
+            for key, value in enumerate(data):
+                # Although the value in the configuration (`data`) is expected to contain 1 or more entries, only
+                # the first validator defined in `validator` property list will be used. It is only defined as a
+                # `list` in `validator` so this logic can understand that the value in `data` attribute should be a
+                # `list`. For example: "pg_hba": [str] in `validator` attribute defines that "pg_hba" in `data`
+                # attribute should contain a list with one or more `str` entries.
+                for v in Schema(validators[0]).validate(value):
+                    yield Result(v.status, v.error,
+                                 path=(str(key) + ("." + v.path if v.path else "")), level=v.level, data=value)
 
     def iter_dict(self) -> Iterator[Result]:
         """Iterate over a :class:`dict` based ``validator`` to validate the corresponding entries in ``data``.
@@ -758,27 +754,26 @@ class Schema(object):
         """
         # One key in `validator` attribute (`key` variable) can be mapped to one or more keys in `data` attribute (`d`
         # variable), depending on the `key` type.
-        if TYPE_CHECKING:  # pragma: no cover
-            assert isinstance(self.validator, dict)
-            assert isinstance(self.data, dict)
-        for key in self.validator.keys():
+        data = cast(Dict[Any, Any], self.data)
+        validators = cast(Dict[Any, Any], self.validator)
+        for key in validators.keys():
             if isinstance(key, AtMostOne) and len(list(self._data_key(key))) > 1:
                 yield Result(False, f"Multiple of {key.args} provided")
                 continue
             for d in self._data_key(key):
-                if d not in self.data and not isinstance(key, Optional):
+                if d not in data and not isinstance(key, Optional):
                     yield Result(False, "is not defined.", path=d)
-                elif d not in self.data and isinstance(key, Optional) and key.default is None:
+                elif d not in data and isinstance(key, Optional) and key.default is None:
                     continue
                 else:
-                    if d not in self.data and isinstance(key, Optional):
-                        self.data[d] = key.default
-                    validator = self.validator[key]
-                    if isinstance(key, (Or, AtMostOne)) and isinstance(self.validator[key], Case):
-                        validator = self.validator[key]._schema[d]
+                    if d not in data and isinstance(key, Optional):
+                        data[d] = key.default
+                    validator = validators[key]
+                    if isinstance(key, (Or, AtMostOne)) and isinstance(validators[key], Case):
+                        validator = validators[key]._schema[d]
                     # In this loop we may be calling a new `Schema` either over an intermediate node in the tree, or
                     # over a leaf node. In the latter case the recursive calls in the given path will finish.
-                    for v in Schema(validator).validate(self.data[d]):
+                    for v in Schema(validator).validate(data[d]):
                         yield Result(v.status, v.error,
                                      path=(d + ("." + v.path if v.path else "")), level=v.level, data=v.data)
 
@@ -818,22 +813,24 @@ class Schema(object):
 
         :yields: keys that should be used to access corresponding value in the ``data`` attribute.
         """
-        # If the key was defined as a `str` object in `validator` attribute, then it is already the final key to access
-        # the `data` dictionary.
-        if isinstance(self.data, dict) and isinstance(key, str):
-            yield key
+        data = cast(Dict[Any, Any], self.data)
+
         # If the key was defined as an `Optional` object in `validator` attribute, then its name is the key to access
         # the `data` dictionary.
-        elif isinstance(key, Optional):
+        if isinstance(key, Optional):
             yield key.name
-        # If the key was defined as an `Or` object in `validator` attribute, then each of its values are the keys to
-        # access the `data` dictionary.
-        elif isinstance(key, Or) and isinstance(self.data, dict):
-            # At least one of the `Or` entries should be available in the `data` dictionary. If we find at least one of
-            # them in `data`, then we return all found entries so the caller method can validate them all.
-            if any([item in self.data for item in key.args]):
+        # If the key was defined as a `str` object in `validator` attribute, then it is already the final key
+        # to access the `data` dictionary.
+        elif isinstance(key, str):
+            yield key
+        # If the key was defined as an `Or` object in `validator` attribute, then each of its values are
+        # the keys to access the `data` dictionary.
+        elif isinstance(key, Or):
+            # At least one of the `Or` entries should be available in the `data` dictionary. If we find at least
+            # one of them in `data`, then we return all found entries so the caller method can validate them all.
+            if any([item in data for item in key.args]):
                 for item in key.args:
-                    if item in self.data:
+                    if item in data:
                         yield item
             # If none of the `Or` entries is available in the `data` dictionary, then we return all entries so the
             # caller method will issue errors that they are all absent.
@@ -842,11 +839,11 @@ class Schema(object):
                     yield item
         # If the key was defined as a `AtMostOne` object in `validator` attribute, then each of its values
         # are the keys to access the `data` dictionary.
-        elif isinstance(key, AtMostOne) and isinstance(self.data, dict):
+        elif isinstance(key, AtMostOne):  # pyright: ignore [reportUnnecessaryIsInstance]
             # Yield back all of the entries from the `data` dictionary, each will be validated and then counted
             # to inform us if we've provided too many
             for item in key.args:
-                if item in self.data:
+                if item in data:
                     yield item