[psutil] Use more strict pyright settings (#15109)

Author: donbarbos

pyrightconfig.stricter.json

@@ -72,7 +72,7 @@
         "stubs/pika",
         "stubs/pony",
         "stubs/protobuf",
-        "stubs/psutil",
+        "stubs/psutil/psutil/__init__.pyi",
         "stubs/psycopg2",
         "stubs/pyasn1",
         "stubs/pycurl",

stubs/psutil/psutil/__init__.pyi

@@ -202,7 +202,7 @@ class Process:
         def cpu_affinity(self, cpus: None = None) -> list[int]: ...
         @overload
         def cpu_affinity(self, cpus: list[int]) -> None: ...
-        def memory_maps(self, grouped: bool = True): ...
+        def memory_maps(self, grouped: bool = True) -> list[Incomplete]: ...
     if sys.platform == "linux":
         def rlimit(self, resource: int, limits: tuple[int, int] | None = ...) -> tuple[int, int]: ...
         def cpu_num(self) -> int: ...

stubs/psutil/psutil/_common.pyi

@@ -1,7 +1,9 @@
 import enum
 import io
+import sys
 import threading
 from _typeshed import ConvertibleToFloat, FileDescriptorOrPath, Incomplete, StrOrBytesPath, SupportsWrite
+from collections import defaultdict
 from collections.abc import Callable
 from socket import AF_INET6 as AF_INET6, AddressFamily, SocketKind
 from typing import BinaryIO, Final, NamedTuple, SupportsIndex, TypeVar, overload
@@ -213,9 +215,7 @@ class addr(NamedTuple):
 
 conn_tmap: dict[str, tuple[list[AddressFamily], list[SocketKind]]]
 
-class Error(Exception):
-    msg: str
-    def __init__(self, msg: str = ...) -> None: ...
+class Error(Exception): ...
 
 class NoSuchProcess(Error):
     pid: int
@@ -242,6 +242,7 @@ class TimeoutExpired(Error):
 
 _P = ParamSpec("_P")
 _R = TypeVar("_R")
+_T = TypeVar("_T")
 
 def usage_percent(used: ConvertibleToFloat, total: float, round_: SupportsIndex | None = None) -> float: ...
 
@@ -257,32 +258,64 @@ def parse_environ_block(data: str) -> dict[str, str]: ...
 def sockfam_to_enum(num: int) -> AddressFamily: ...
 def socktype_to_enum(num: int) -> SocketKind: ...
 @overload
-def conn_to_ntuple(fd: int, fam: int, type_: int, laddr, raddr, status: str, status_map, pid: int) -> sconn: ...
+def conn_to_ntuple(
+    fd: int,
+    fam: int,
+    type_: int,
+    laddr: addr | tuple[str, int] | tuple[()],
+    raddr: addr | tuple[str, int] | tuple[()],
+    status: int | str,
+    status_map: dict[int, str] | dict[str, str],
+    pid: int,
+) -> sconn: ...
 @overload
-def conn_to_ntuple(fd: int, fam: int, type_: int, laddr, raddr, status: str, status_map, pid: None = None) -> pconn: ...
+def conn_to_ntuple(
+    fd: int,
+    fam: int,
+    type_: int,
+    laddr: addr | tuple[str, int] | tuple[()],
+    raddr: addr | tuple[str, int] | tuple[()],
+    status: int | str,
+    status_map: dict[int, str] | dict[str, str],
+    pid: None = None,
+) -> pconn: ...
 def deprecated_method(replacement: str) -> Callable[[Callable[_P, _R]], Callable[_P, _R]]: ...
 
 class _WrapNumbers:
     lock: threading.Lock
-    cache: dict[Incomplete, Incomplete]
-    reminders: dict[Incomplete, Incomplete]
-    reminder_keys: dict[Incomplete, Incomplete]
+    cache: dict[str, dict[str, tuple[int, ...]]]
+    reminders: dict[str, defaultdict[Incomplete, int]]
+    reminder_keys: dict[str, defaultdict[Incomplete, set[Incomplete]]]
     def __init__(self) -> None: ...
-    def run(self, input_dict, name): ...
-    def cache_clear(self, name=None) -> None: ...
-    def cache_info(self) -> tuple[dict[Incomplete, Incomplete], dict[Incomplete, Incomplete], dict[Incomplete, Incomplete]]: ...
-
-def wrap_numbers(input_dict, name: str): ...
+    def run(self, input_dict: dict[str, tuple[int, ...]], name: str) -> dict[str, tuple[int, ...]]: ...
+    def cache_clear(self, name: str | None = None) -> None: ...
+    def cache_info(
+        self,
+    ) -> tuple[
+        dict[str, dict[str, tuple[int, ...]]],
+        dict[str, defaultdict[Incomplete, int]],
+        dict[str, defaultdict[Incomplete, set[Incomplete]]],
+    ]: ...
+
+def wrap_numbers(input_dict: dict[str, tuple[int, ...]], name: str) -> dict[str, tuple[int, ...]]: ...
 def open_binary(fname: FileDescriptorOrPath) -> BinaryIO: ...
 def open_text(fname: FileDescriptorOrPath) -> io.TextIOWrapper: ...
-def cat(fname: FileDescriptorOrPath, fallback=..., _open=...): ...
-def bcat(fname: FileDescriptorOrPath, fallback=...): ...
+@overload
+def cat(fname: FileDescriptorOrPath, _open: Callable[[FileDescriptorOrPath], io.TextIOWrapper] = ...) -> str: ...
+@overload
+def cat(
+    fname: FileDescriptorOrPath, fallback: _T = ..., _open: Callable[[FileDescriptorOrPath], io.TextIOWrapper] = ...
+) -> str | _T: ...
+@overload
+def bcat(fname: FileDescriptorOrPath) -> str: ...
+@overload
+def bcat(fname: FileDescriptorOrPath, fallback: _T = ...) -> str | _T: ...
 def bytes2human(n: int, format: str = "%(value).1f%(symbol)s") -> str: ...
 def get_procfs_path() -> str: ...
 def decode(s: bytes) -> str: ...
-def term_supports_colors(file: SupportsWrite[str] = ...) -> bool: ...
+def term_supports_colors(file: SupportsWrite[str] = sys.stdout) -> bool: ...
 def hilite(s: str, color: str | None = None, bold: bool = False) -> str: ...
-def print_color(s: str, color: str | None = None, bold: bool = False, file: SupportsWrite[str] = ...) -> None: ...
+def print_color(s: str, color: str | None = None, bold: bool = False, file: SupportsWrite[str] = sys.stdout) -> None: ...
 def debug(msg: str | Exception) -> None: ...
 
 __all__ = [

stubs/psutil/psutil/_psosx.pyi

@@ -108,10 +108,10 @@ if sys.platform == "darwin":
         def num_ctx_switches(self) -> _common.pctxsw: ...
         def num_threads(self) -> int: ...
         def open_files(self) -> list[_common.popenfile]: ...
-        def net_connections(self, kind: str = "inet"): ...
+        def net_connections(self, kind: str = "inet") -> list[_common.pconn]: ...
         def num_fds(self) -> int: ...
         def wait(self, timeout: float | None = None) -> int | None: ...
-        def nice_get(self): ...
-        def nice_set(self, value): ...
+        def nice_get(self) -> int: ...
+        def nice_set(self, value: int) -> None: ...
         def status(self) -> str: ...
         def threads(self) -> list[_common.pthread]: ...

stubs/psutil/psutil/_pswindows.pyi

@@ -3,9 +3,10 @@ import sys
 if sys.platform == "win32":
     import enum
     from _typeshed import Incomplete
-    from collections.abc import Iterable, Iterator
+    from collections.abc import Callable, Iterable, Iterator
     from signal import Signals
-    from typing import Final, Literal, NamedTuple, TypedDict, overload, type_check_only
+    from typing import Final, Literal, NamedTuple, TypedDict, TypeVar, overload, type_check_only
+    from typing_extensions import ParamSpec
 
     from psutil import _psutil_windows
     from psutil._common import (
@@ -127,6 +128,9 @@ if sys.platform == "win32":
         other_count: int
         other_bytes: int
 
+    _P = ParamSpec("_P")
+    _R = TypeVar("_R")
+
     def convert_dos_path(s: str) -> str: ...
     def getpagesize() -> int: ...
     def virtual_memory() -> svmem: ...
@@ -186,11 +190,11 @@ if sys.platform == "win32":
 
     def is_permission_err(exc: OSError) -> bool: ...
     @overload
-    def convert_oserror(exc: PermissionError, pid=None, name=None) -> AccessDenied: ...
+    def convert_oserror(exc: PermissionError, pid: int | None = None, name: str | None = None) -> AccessDenied: ...
     @overload
-    def convert_oserror(exc: OSError, pid=None, name=None) -> AccessDenied | NoSuchProcess: ...
-    def wrap_exceptions(fun): ...
-    def retry_error_partial_copy(fun): ...
+    def convert_oserror(exc: OSError, pid: int | None = None, name: str | None = None) -> AccessDenied | NoSuchProcess: ...
+    def wrap_exceptions(fun: Callable[_P, _R]) -> Callable[_P, _R]: ...
+    def retry_error_partial_copy(fun: Callable[_P, _R]) -> Callable[_P, _R]: ...
 
     class Process:
         __slots__ = ["_cache", "_name", "_ppid", "pid"]
@@ -220,7 +224,7 @@ if sys.platform == "win32":
         def open_files(self) -> list[_common.popenfile]: ...
         def net_connections(self, kind: str = "inet") -> list[_common.pconn]: ...
         def nice_get(self) -> Priority: ...
-        def nice_set(self, value) -> None: ...
+        def nice_set(self, value: int) -> None: ...
         def ionice_get(self) -> IOPriority: ...
         def ionice_set(self, ioclass: int, value: None) -> None: ...
         def io_counters(self) -> pio: ...