Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion pyrightconfig.stricter.json
Original file line number Diff line number Diff line change
Expand Up @@ -72,7 +72,7 @@
"stubs/pika",
"stubs/pony",
"stubs/protobuf",
"stubs/psutil",
"stubs/psutil/psutil/__init__.pyi",
"stubs/psycopg2",
"stubs/pyasn1",
"stubs/pycurl",
Expand Down
2 changes: 1 addition & 1 deletion stubs/psutil/psutil/__init__.pyi
Original file line number Diff line number Diff line change
Expand Up @@ -199,7 +199,7 @@ class Process:
def io_counters(self) -> pio: ...
def ionice(self, ioclass: int | None = None, value: int | None = None) -> pionice: ...
def cpu_affinity(self, cpus: list[int] | None = None) -> 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: ...
Expand Down
67 changes: 50 additions & 17 deletions stubs/psutil/psutil/_common.pyi
Original file line number Diff line number Diff line change
@@ -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
Expand Down Expand Up @@ -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
Expand All @@ -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: ...

Expand All @@ -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__ = [
Expand Down
6 changes: 3 additions & 3 deletions stubs/psutil/psutil/_psosx.pyi
Original file line number Diff line number Diff line change
Expand Up @@ -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]: ...
18 changes: 11 additions & 7 deletions stubs/psutil/psutil/_pswindows.pyi
Original file line number Diff line number Diff line change
Expand Up @@ -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 (
Expand Down Expand Up @@ -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: ...
Expand Down Expand Up @@ -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"]
Expand Down Expand Up @@ -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: ...
Expand Down