Enable pyi-stubs for several modules and packages

_bisect, _bootlocale, _compat_pickle, _msi, _py_abc, _sitebuiltins,
_tracemalloc, _winapi, _winreg, aifc, atomicwrites, base64, bz2,
chardet, compileall, ftplib, hmac, lzma, mmap, msilib, pkg_resources,
random, sunau, tracemalloc, tzlocal, urlparse, waitress, wave,
winreg, wsgiref, xmlrpc

GitOrigin-RevId: f0cdd560c24199b2f16dce094b0be2240950fb13
This commit is contained in:
Semyon Proshev
2020-06-02 22:37:57 +03:00
committed by intellij-monorepo-bot
parent f690961be8
commit 425b711c8a
70 changed files with 3873 additions and 0 deletions
@@ -0,0 +1,99 @@
import sys
from typing import Optional, Type, Union, Tuple, Any
from types import TracebackType
_KeyType = Union[HKEYType, int]
def CloseKey(__hkey: _KeyType): ...
def ConnectRegistry(__computer_name: Optional[str], __key: _KeyType) -> HKEYType: ...
def CreateKey(__key: _KeyType, __sub_key: Optional[str]) -> HKEYType: ...
def CreateKeyEx(key: _KeyType, sub_key: Optional[str], reserved: int = ..., access: int = ...) -> HKEYType: ...
def DeleteKey(__key: _KeyType, __sub_key: str) -> None: ...
def DeleteKeyEx(key: _KeyType, sub_key: str, access: int = ..., reserved: int = ...) -> None: ...
def DeleteValue(__key: _KeyType, __value: str) -> None: ...
def EnumKey(__key: _KeyType, __index: int) -> str: ...
def EnumValue(__key: _KeyType, __index: int) -> Tuple[str, Any, int]: ...
def ExpandEnvironmentStrings(__str: str) -> str: ...
def FlushKey(__key: _KeyType) -> None: ...
def LoadKey(__key: _KeyType, __sub_key: str, __file_name: str) -> None: ...
def OpenKey(key: _KeyType, sub_key: str, reserved: int = ..., access: int = ...) -> HKEYType: ...
def OpenKeyEx(key: _KeyType, sub_key: str, reserved: int = ..., access: int = ...) -> HKEYType: ...
def QueryInfoKey(__key: _KeyType) -> Tuple[int, int, int]: ...
def QueryValue(__key: _KeyType, __sub_key: Optional[str]) -> str: ...
def QueryValueEx(__key: _KeyType, __name: str) -> Tuple[Any, int]: ...
def SaveKey(__key: _KeyType, __file_name: str) -> None: ...
def SetValue(__key: _KeyType, __sub_key: str, __type: int, __value: str) -> None: ...
def SetValueEx(__key: _KeyType, __value_name: str, __reserved: Any, __type: int, __value: str) -> None: ... # reserved is ignored
def DisableReflectionKey(__key: _KeyType) -> None: ...
def EnableReflectionKey(__key: _KeyType) -> None: ...
def QueryReflectionKey(__key: _KeyType) -> bool: ...
HKEY_CLASSES_ROOT: int
HKEY_CURRENT_USER: int
HKEY_LOCAL_MACHINE: int
HKEY_USERS: int
HKEY_PERFORMANCE_DATA: int
HKEY_CURRENT_CONFIG: int
HKEY_DYN_DATA: int
KEY_ALL_ACCESS: int
KEY_WRITE: int
KEY_READ: int
KEY_EXECUTE: int
KEY_QUERY_VALUE: int
KEY_SET_VALUE: int
KEY_CREATE_SUB_KEY: int
KEY_ENUMERATE_SUB_KEYS: int
KEY_NOTIFY: int
KEY_CREATE_LINK: int
KEY_WOW64_64KEY: int
KEY_WOW64_32KEY: int
REG_BINARY: int
REG_DWORD: int
REG_DWORD_LITTLE_ENDIAN: int
REG_DWORD_BIG_ENDIAN: int
REG_EXPAND_SZ: int
REG_LINK: int
REG_MULTI_SZ: int
REG_NONE: int
if sys.version_info >= (3, 6):
REG_QWORD: int
REG_QWORD_LITTLE_ENDIAN: int
REG_RESOURCE_LIST: int
REG_FULL_RESOURCE_DESCRIPTOR: int
REG_RESOURCE_REQUIREMENTS_LIST: int
REG_SZ: int
REG_CREATED_NEW_KEY: int # undocumented
REG_LEGAL_CHANGE_FILTER: int # undocumented
REG_LEGAL_OPTION: int # undocumented
REG_NOTIFY_CHANGE_ATTRIBUTES: int # undocumented
REG_NOTIFY_CHANGE_LAST_SET: int # undocumented
REG_NOTIFY_CHANGE_NAME: int # undocumented
REG_NOTIFY_CHANGE_SECURITY: int # undocumented
REG_NO_LAZY_FLUSH: int # undocumented
REG_OPENED_EXISTING_KEY: int # undocumented
REG_OPTION_BACKUP_RESTORE: int # undocumented
REG_OPTION_CREATE_LINK: int # undocumented
REG_OPTION_NON_VOLATILE: int # undocumented
REG_OPTION_OPEN_LINK: int # undocumented
REG_OPTION_RESERVED: int # undocumented
REG_OPTION_VOLATILE: int # undocumented
REG_REFRESH_HIVE: int # undocumented
REG_WHOLE_HIVE_VOLATILE: int # undocumented
error = OSError
# Though this class has a __name__ of PyHKEY, it's exposed as HKEYType for some reason
class HKEYType:
def __bool__(self) -> bool: ...
def __int__(self) -> int: ...
def __enter__(self) -> HKEYType: ...
def __exit__(self, exc_type: Optional[Type[BaseException]], exc_val: Optional[BaseException], exc_tb: Optional[TracebackType]) -> Optional[bool]: ...
def Close(self) -> None: ...
def Detach(self) -> int: ...
@@ -0,0 +1,19 @@
# Stubs for compileall (Python 2)
from typing import Any, Optional, Pattern, Union
_Path = Union[str, bytes]
# rx can be any object with a 'search' method; once we have Protocols we can change the type
def compile_dir(
dir: _Path,
maxlevels: int = ...,
ddir: Optional[_Path] = ...,
force: bool = ...,
rx: Optional[Pattern[Any]] = ...,
quiet: int = ...,
) -> int: ...
def compile_file(
fullname: _Path, ddir: Optional[_Path] = ..., force: bool = ..., rx: Optional[Pattern[Any]] = ..., quiet: int = ...,
) -> int: ...
def compile_path(skip_curdir: bool = ..., maxlevels: int = ..., force: bool = ..., quiet: int = ...) -> int: ...
@@ -0,0 +1,77 @@
# Stubs for random
# Ron Murawski <ron@horizonchess.com>
# Updated by Jukka Lehtosalo
# based on https://docs.python.org/2/library/random.html
# ----- random classes -----
import _random
from typing import AbstractSet, Any, Callable, Iterator, List, Protocol, Sequence, TypeVar, Union, overload
_T = TypeVar("_T")
_T_co = TypeVar('_T_co', covariant=True)
class _Sampleable(Protocol[_T_co]):
def __iter__(self) -> Iterator[_T_co]: ...
def __len__(self) -> int: ...
class Random(_random.Random):
def __init__(self, x: object = ...) -> None: ...
def seed(self, x: object = ...) -> None: ...
def getstate(self) -> _random._State: ...
def setstate(self, state: _random._State) -> None: ...
def jumpahead(self, n: int) -> None: ...
def getrandbits(self, k: int) -> int: ...
@overload
def randrange(self, stop: int) -> int: ...
@overload
def randrange(self, start: int, stop: int, step: int = ...) -> int: ...
def randint(self, a: int, b: int) -> int: ...
def choice(self, seq: Sequence[_T]) -> _T: ...
def shuffle(self, x: List[Any], random: Callable[[], None] = ...) -> None: ...
def sample(self, population: _Sampleable[_T], k: int) -> List[_T]: ...
def random(self) -> float: ...
def uniform(self, a: float, b: float) -> float: ...
def triangular(self, low: float = ..., high: float = ..., mode: float = ...) -> float: ...
def betavariate(self, alpha: float, beta: float) -> float: ...
def expovariate(self, lambd: float) -> float: ...
def gammavariate(self, alpha: float, beta: float) -> float: ...
def gauss(self, mu: float, sigma: float) -> float: ...
def lognormvariate(self, mu: float, sigma: float) -> float: ...
def normalvariate(self, mu: float, sigma: float) -> float: ...
def vonmisesvariate(self, mu: float, kappa: float) -> float: ...
def paretovariate(self, alpha: float) -> float: ...
def weibullvariate(self, alpha: float, beta: float) -> float: ...
# SystemRandom is not implemented for all OS's; good on Windows & Linux
class SystemRandom(Random):
...
# ----- random function stubs -----
def seed(x: object = ...) -> None: ...
def getstate() -> object: ...
def setstate(state: object) -> None: ...
def jumpahead(n: int) -> None: ...
def getrandbits(k: int) -> int: ...
@overload
def randrange(stop: int) -> int: ...
@overload
def randrange(start: int, stop: int, step: int = ...) -> int: ...
def randint(a: int, b: int) -> int: ...
def choice(seq: Sequence[_T]) -> _T: ...
def shuffle(x: List[Any], random: Callable[[], float] = ...) -> None: ...
def sample(population: _Sampleable[_T], k: int) -> List[_T]: ...
def random() -> float: ...
def uniform(a: float, b: float) -> float: ...
def triangular(low: float = ..., high: float = ...,
mode: float = ...) -> float: ...
def betavariate(alpha: float, beta: float) -> float: ...
def expovariate(lambd: float) -> float: ...
def gammavariate(alpha: float, beta: float) -> float: ...
def gauss(mu: float, sigma: float) -> float: ...
def lognormvariate(mu: float, sigma: float) -> float: ...
def normalvariate(mu: float, sigma: float) -> float: ...
def vonmisesvariate(mu: float, kappa: float) -> float: ...
def paretovariate(alpha: float) -> float: ...
def weibullvariate(alpha: float, beta: float) -> float: ...
@@ -0,0 +1,66 @@
# Stubs for urlparse (Python 2)
from typing import AnyStr, Dict, List, NamedTuple, Tuple, Sequence, Union, overload, Optional
_String = Union[str, unicode]
uses_relative: List[str]
uses_netloc: List[str]
uses_params: List[str]
non_hierarchical: List[str]
uses_query: List[str]
uses_fragment: List[str]
scheme_chars: str
MAX_CACHE_SIZE: int
def clear_cache() -> None: ...
class ResultMixin(object):
@property
def username(self) -> Optional[_String]: ...
@property
def password(self) -> Optional[_String]: ...
@property
def hostname(self) -> Optional[_String]: ...
@property
def port(self) -> Optional[int]: ...
class _SplitResult(NamedTuple):
scheme: _String
netloc: _String
path: _String
query: _String
fragment: _String
class SplitResult(_SplitResult, ResultMixin):
def geturl(self) -> _String: ...
class _ParseResult(NamedTuple):
scheme: _String
netloc: _String
path: _String
params: _String
query: _String
fragment: _String
class ParseResult(_ParseResult, ResultMixin):
def geturl(self) -> _String: ...
def urlparse(url: _String, scheme: _String = ...,
allow_fragments: bool = ...) -> ParseResult: ...
def urlsplit(url: _String, scheme: _String = ...,
allow_fragments: bool = ...) -> SplitResult: ...
@overload
def urlunparse(data: Tuple[_String, _String, _String, _String, _String, _String]) -> _String: ...
@overload
def urlunparse(data: Sequence[_String]) -> _String: ...
@overload
def urlunsplit(data: Tuple[_String, _String, _String, _String, _String]) -> _String: ...
@overload
def urlunsplit(data: Sequence[_String]) -> _String: ...
def urljoin(base: _String, url: _String,
allow_fragments: bool = ...) -> _String: ...
def urldefrag(url: AnyStr) -> Tuple[AnyStr, AnyStr]: ...
def unquote(s: AnyStr) -> AnyStr: ...
def parse_qs(qs: AnyStr, keep_blank_values: bool = ...,
strict_parsing: bool = ...) -> Dict[AnyStr, List[AnyStr]]: ...
def parse_qsl(qs: AnyStr, keep_blank_values: int = ...,
strict_parsing: bool = ...) -> List[Tuple[AnyStr, AnyStr]]: ...
@@ -0,0 +1,9 @@
"""Stub file for the '_bisect' module."""
from typing import Optional, Sequence, MutableSequence, TypeVar
_T = TypeVar('_T')
def bisect_left(a: Sequence[_T], x: _T, lo: int = ..., hi: Optional[int] = ...) -> int: ...
def bisect_right(a: Sequence[_T], x: _T, lo: int = ..., hi: Optional[int] = ...) -> int: ...
def insort_left(a: MutableSequence[_T], x: _T, lo: int = ..., hi: Optional[int] = ...) -> None: ...
def insort_right(a: MutableSequence[_T], x: _T, lo: int = ..., hi: Optional[int] = ...) -> None: ...
@@ -0,0 +1,62 @@
import sys
from typing import List, Optional, Union
if sys.platform == 'win32':
# Actual typename View, not exposed by the implementation
class _View:
def Execute(self, params: Optional[_Record] = ...) -> None: ...
def GetColumnInfo(self, kind: int) -> _Record: ...
def Fetch(self) -> _Record: ...
def Modify(self, mode: int, record: _Record) -> None: ...
def Close(self) -> None: ...
# Don't exist at runtime
__new__: None # type: ignore
__init__: None # type: ignore
# Actual typename Summary, not exposed by the implementation
class _Summary:
def GetProperty(self, propid: int) -> Optional[Union[str, bytes]]: ...
def GetPropertyCount(self) -> int: ...
def SetProperty(self, propid: int, value: Union[str, bytes]) -> None: ...
def Persist(self) -> None: ...
# Don't exist at runtime
__new__: None # type: ignore
__init__: None # type: ignore
# Actual typename Database, not exposed by the implementation
class _Database:
def OpenView(self, sql: str) -> _View: ...
def Commit(self) -> None: ...
def GetSummaryInformation(self, updateCount: int) -> _Summary: ...
def Close(self) -> None: ...
# Don't exist at runtime
__new__: None # type: ignore
__init__: None # type: ignore
# Actual typename Record, not exposed by the implementation
class _Record:
def GetFieldCount(self) -> int: ...
def GetInteger(self, field: int) -> int: ...
def GetString(self, field: int) -> str: ...
def SetString(self, field: int, str: str) -> None: ...
def SetStream(self, field: int, stream: str) -> None: ...
def SetInteger(self, field: int, int: int) -> None: ...
def ClearData(self) -> None: ...
# Don't exist at runtime
__new__: None # type: ignore
__init__: None # type: ignore
def UuidCreate() -> str: ...
def FCICreate(cabname: str, files: List[str]) -> None: ...
def OpenDatabase(name: str, flags: int) -> _Database: ...
def CreateRecord(count: int) -> _Record: ...
@@ -0,0 +1,86 @@
from typing import Union, IO, Optional, Type, NamedTuple, List, Tuple, Any, Text, overload
from typing_extensions import Literal
from types import TracebackType
import sys
class Error(Exception): ...
class _aifc_params(NamedTuple):
nchannels: int
sampwidth: int
framerate: int
nframes: int
comptype: bytes
compname: bytes
_File = Union[Text, IO[bytes]]
_Marker = Tuple[int, int, bytes]
class Aifc_read:
def __init__(self, f: _File) -> None: ...
if sys.version_info >= (3, 4):
def __enter__(self) -> Aifc_read: ...
def __exit__(self, exc_type: Optional[Type[BaseException]], exc_val: Optional[BaseException],
exc_tb: Optional[TracebackType]) -> None: ...
def initfp(self, file: IO[bytes]) -> None: ...
def getfp(self) -> IO[bytes]: ...
def rewind(self) -> None: ...
def close(self) -> None: ...
def tell(self) -> int: ...
def getnchannels(self) -> int: ...
def getnframes(self) -> int: ...
def getsampwidth(self) -> int: ...
def getframerate(self) -> int: ...
def getcomptype(self) -> bytes: ...
def getcompname(self) -> bytes: ...
def getparams(self) -> _aifc_params: ...
def getmarkers(self) -> Optional[List[_Marker]]: ...
def getmark(self, id: int) -> _Marker: ...
def setpos(self, pos: int) -> None: ...
def readframes(self, nframes: int) -> bytes: ...
class Aifc_write:
def __init__(self, f: _File) -> None: ...
def __del__(self) -> None: ...
if sys.version_info >= (3, 4):
def __enter__(self) -> Aifc_write: ...
def __exit__(self, exc_type: Optional[Type[BaseException]], exc_val: Optional[BaseException],
exc_tb: Optional[TracebackType]) -> None: ...
def initfp(self, file: IO[bytes]) -> None: ...
def aiff(self) -> None: ...
def aifc(self) -> None: ...
def setnchannels(self, nchannels: int) -> None: ...
def getnchannels(self) -> int: ...
def setsampwidth(self, sampwidth: int) -> None: ...
def getsampwidth(self) -> int: ...
def setframerate(self, framerate: int) -> None: ...
def getframerate(self) -> int: ...
def setnframes(self, nframes: int) -> None: ...
def getnframes(self) -> int: ...
def setcomptype(self, comptype: bytes, compname: bytes) -> None: ...
def getcomptype(self) -> bytes: ...
def getcompname(self) -> bytes: ...
def setparams(self, params: Tuple[int, int, int, int, bytes, bytes]) -> None: ...
def getparams(self) -> _aifc_params: ...
def setmark(self, id: int, pos: int, name: bytes) -> None: ...
def getmark(self, id: int) -> _Marker: ...
def getmarkers(self) -> Optional[List[_Marker]]: ...
def tell(self) -> int: ...
def writeframesraw(self, data: Any) -> None: ... # Actual type for data is Buffer Protocol
def writeframes(self, data: Any) -> None: ...
def close(self) -> None: ...
@overload
def open(f: _File, mode: Literal["r", "rb"]) -> Aifc_read: ...
@overload
def open(f: _File, mode: Literal["w", "wb"]) -> Aifc_write: ...
@overload
def open(f: _File, mode: Optional[str] = ...) -> Any: ...
if sys.version_info < (3, 9):
@overload
def openfp(f: _File, mode: Literal["r", "rb"]) -> Aifc_read: ...
@overload
def openfp(f: _File, mode: Literal["w", "wb"]) -> Aifc_write: ...
@overload
def openfp(f: _File, mode: Optional[str] = ...) -> Any: ...
@@ -0,0 +1,40 @@
# Stubs for base64
from typing import IO, Optional, Text, Union
import sys
if sys.version_info < (3,):
_encodable = Union[bytes, unicode]
_decodable = Union[bytes, unicode]
else:
_encodable = bytes
_decodable = Union[bytes, str]
def b64encode(s: _encodable, altchars: Optional[bytes] = ...) -> bytes: ...
def b64decode(s: _decodable, altchars: Optional[bytes] = ...,
validate: bool = ...) -> bytes: ...
def standard_b64encode(s: _encodable) -> bytes: ...
def standard_b64decode(s: _decodable) -> bytes: ...
def urlsafe_b64encode(s: _encodable) -> bytes: ...
def urlsafe_b64decode(s: _decodable) -> bytes: ...
def b32encode(s: _encodable) -> bytes: ...
def b32decode(s: _decodable, casefold: bool = ...,
map01: Optional[bytes] = ...) -> bytes: ...
def b16encode(s: _encodable) -> bytes: ...
def b16decode(s: _decodable, casefold: bool = ...) -> bytes: ...
if sys.version_info >= (3, 4):
def a85encode(b: _encodable, *, foldspaces: bool = ..., wrapcol: int = ...,
pad: bool = ..., adobe: bool = ...) -> bytes: ...
def a85decode(b: _decodable, *, foldspaces: bool = ...,
adobe: bool = ..., ignorechars: Union[str, bytes] = ...) -> bytes: ...
def b85encode(b: _encodable, pad: bool = ...) -> bytes: ...
def b85decode(b: _decodable) -> bytes: ...
def decode(input: IO[bytes], output: IO[bytes]) -> None: ...
def encode(input: IO[bytes], output: IO[bytes]) -> None: ...
if sys.version_info >= (3,):
def encodebytes(s: bytes) -> bytes: ...
def decodebytes(s: bytes) -> bytes: ...
if sys.version_info < (3, 9):
def encodestring(s: bytes) -> bytes: ...
def decodestring(s: bytes) -> bytes: ...
@@ -0,0 +1,80 @@
import io
import sys
from os.path import _PathType
from typing import IO, Any, Optional, TextIO, Union, overload, TypeVar
if sys.version_info >= (3, 8):
from typing import Literal
else:
from typing_extensions import Literal
_PathOrFile = Union[_PathType, IO[bytes]]
_T = TypeVar("_T")
def compress(data: bytes, compresslevel: int = ...) -> bytes: ...
def decompress(data: bytes) -> bytes: ...
if sys.version_info >= (3, 3):
_OpenBinaryMode = Literal["r", "rb", "w", "wb", "x", "xb", "a", "ab"]
_OpenTextMode = Literal["rt", "wt", "xt", "at"]
@overload
def open(
filename: _PathOrFile,
mode: _OpenBinaryMode = ...,
compresslevel: int = ...,
encoding: None = ...,
errors: None = ...,
newline: None = ...,
) -> BZ2File: ...
@overload
def open(
filename: _PathType,
mode: _OpenTextMode,
compresslevel: int = ...,
encoding: Optional[str] = ...,
errors: Optional[str] = ...,
newline: Optional[str] = ...,
) -> TextIO: ...
@overload
def open(
filename: _PathOrFile,
mode: str,
compresslevel: int = ...,
encoding: Optional[str] = ...,
errors: Optional[str] = ...,
newline: Optional[str] = ...,
) -> Union[BZ2File, TextIO]: ...
class BZ2File(io.BufferedIOBase, IO[bytes]):
def __enter__(self: _T) -> _T: ...
if sys.version_info >= (3, 9):
def __init__(self,
filename: _PathOrFile,
mode: str = ...,
*,
compresslevel: int = ...) -> None: ...
else:
def __init__(self,
filename: _PathOrFile,
mode: str = ...,
buffering: Optional[Any] = ...,
compresslevel: int = ...) -> None: ...
class BZ2Compressor(object):
def __init__(self, compresslevel: int = ...) -> None: ...
def compress(self, data: bytes) -> bytes: ...
def flush(self) -> bytes: ...
class BZ2Decompressor(object):
if sys.version_info >= (3, 5):
def decompress(self, data: bytes, max_length: int = ...) -> bytes: ...
else:
def decompress(self, data: bytes) -> bytes: ...
if sys.version_info >= (3, 3):
@property
def eof(self) -> bool: ...
if sys.version_info >= (3, 5):
@property
def needs_input(self) -> bool: ...
@property
def unused_data(self) -> bytes: ...
@@ -0,0 +1,174 @@
import sys
from typing import (
Any,
BinaryIO,
Callable,
Dict,
Generic,
Iterable,
Iterator,
List,
Optional,
Protocol,
Text,
TextIO,
Tuple,
Type,
TypeVar,
Union,
)
from types import TracebackType
from socket import socket
from ssl import SSLContext
_T = TypeVar("_T")
_IntOrStr = Union[int, Text]
MSG_OOB: int
FTP_PORT: int
MAXLINE: int
CRLF: str
if sys.version_info >= (3,):
B_CRLF: bytes
class Error(Exception): ...
class error_reply(Error): ...
class error_temp(Error): ...
class error_perm(Error): ...
class error_proto(Error): ...
all_errors = Tuple[Type[Exception], ...]
class _Readable(Protocol):
def read(self, __length: int) -> bytes: ...
class _ReadLineable(Protocol):
def readline(self, _length: int) -> bytes: ...
class FTP:
debugging: int
# Note: This is technically the type that's passed in as the host argument. But to make it easier in Python 2 we
# accept Text but return str.
host: str
port: int
maxline: int
sock: Optional[socket]
welcome: Optional[str]
passiveserver: int
timeout: int
af: int
lastresp: str
if sys.version_info >= (3,):
file: Optional[TextIO]
encoding: str
def __enter__(self: _T) -> _T: ...
def __exit__(
self, exc_type: Optional[Type[BaseException]], exc_val: Optional[BaseException], exc_tb: Optional[TracebackType]
) -> None: ...
else:
file: Optional[BinaryIO]
if sys.version_info >= (3, 3):
source_address: Optional[Tuple[str, int]]
def __init__(
self,
host: Text = ...,
user: Text = ...,
passwd: Text = ...,
acct: Text = ...,
timeout: float = ...,
source_address: Optional[Tuple[str, int]] = ...,
) -> None: ...
def connect(
self, host: Text = ..., port: int = ..., timeout: float = ..., source_address: Optional[Tuple[str, int]] = ...
) -> str: ...
else:
def __init__(
self, host: Text = ..., user: Text = ..., passwd: Text = ..., acct: Text = ..., timeout: float = ...
) -> None: ...
def connect(self, host: Text = ..., port: int = ..., timeout: float = ...) -> str: ...
def getwelcome(self) -> str: ...
def set_debuglevel(self, level: int) -> None: ...
def debug(self, level: int) -> None: ...
def set_pasv(self, val: Union[bool, int]) -> None: ...
def sanitize(self, s: Text) -> str: ...
def putline(self, line: Text) -> None: ...
def putcmd(self, line: Text) -> None: ...
def getline(self) -> str: ...
def getmultiline(self) -> str: ...
def getresp(self) -> str: ...
def voidresp(self) -> str: ...
def abort(self) -> str: ...
def sendcmd(self, cmd: Text) -> str: ...
def voidcmd(self, cmd: Text) -> str: ...
def sendport(self, host: Text, port: int) -> str: ...
def sendeprt(self, host: Text, port: int) -> str: ...
def makeport(self) -> socket: ...
def makepasv(self) -> Tuple[str, int]: ...
def login(self, user: Text = ..., passwd: Text = ..., acct: Text = ...) -> str: ...
# In practice, `rest` rest can actually be anything whose str() is an integer sequence, so to make it simple we allow integers.
def ntransfercmd(self, cmd: Text, rest: Optional[_IntOrStr] = ...) -> Tuple[socket, int]: ...
def transfercmd(self, cmd: Text, rest: Optional[_IntOrStr] = ...) -> socket: ...
def retrbinary(
self, cmd: Text, callback: Callable[[bytes], Any], blocksize: int = ..., rest: Optional[_IntOrStr] = ...
) -> str: ...
def storbinary(
self,
cmd: Text,
fp: _Readable,
blocksize: int = ...,
callback: Optional[Callable[[bytes], Any]] = ...,
rest: Optional[_IntOrStr] = ...,
) -> str: ...
def retrlines(self, cmd: Text, callback: Optional[Callable[[str], Any]] = ...) -> str: ...
def storlines(self, cmd: Text, fp: _ReadLineable, callback: Optional[Callable[[bytes], Any]] = ...) -> str: ...
def acct(self, password: Text) -> str: ...
def nlst(self, *args: Text) -> List[str]: ...
# Technically only the last arg can be a Callable but ...
def dir(self, *args: Union[str, Callable[[str], None]]) -> None: ...
if sys.version_info >= (3, 3):
def mlsd(self, path: Text = ..., facts: Iterable[str] = ...) -> Iterator[Tuple[str, Dict[str, str]]]: ...
def rename(self, fromname: Text, toname: Text) -> str: ...
def delete(self, filename: Text) -> str: ...
def cwd(self, dirname: Text) -> str: ...
def size(self, filename: Text) -> Optional[int]: ...
def mkd(self, dirname: Text) -> str: ...
def rmd(self, dirname: Text) -> str: ...
def pwd(self) -> str: ...
def quit(self) -> str: ...
def close(self) -> None: ...
class FTP_TLS(FTP):
def __init__(
self,
host: Text = ...,
user: Text = ...,
passwd: Text = ...,
acct: Text = ...,
keyfile: Optional[str] = ...,
certfile: Optional[str] = ...,
context: Optional[SSLContext] = ...,
timeout: float = ...,
source_address: Optional[Tuple[str, int]] = ...,
) -> None: ...
ssl_version: int
keyfile: Optional[str]
certfile: Optional[str]
context: SSLContext
def login(self, user: Text = ..., passwd: Text = ..., acct: Text = ..., secure: bool = ...) -> str: ...
def auth(self) -> str: ...
def prot_p(self) -> str: ...
def prot_c(self) -> str: ...
if sys.version_info >= (3, 3):
def ccc(self) -> str: ...
if sys.version_info < (3,):
class Netrc:
def __init__(self, filename: Optional[Text] = ...) -> None: ...
def get_hosts(self) -> List[str]: ...
def get_account(self, host: Text) -> Tuple[Optional[str], Optional[str], Optional[str]]: ...
def get_macros(self) -> List[str]: ...
def get_macro(self, macro: Text) -> Tuple[str, ...]: ...
@@ -0,0 +1,46 @@
# Stubs for hmac
from typing import Any, Callable, Optional, Union, overload, AnyStr, overload
from types import ModuleType
import sys
_B = Union[bytes, bytearray]
# TODO more precise type for object of hashlib
_Hash = Any
digest_size: None
if sys.version_info >= (3, 8):
_DigestMod = Union[str, Callable[[], _Hash], ModuleType]
# In reality digestmod has a default value, but the function always throws an error
# if the argument is not given, so we pretend it is a required argument.
@overload
def new(key: _B, msg: Optional[_B], digestmod: _DigestMod) -> HMAC: ...
@overload
def new(key: _B, *, digestmod: _DigestMod) -> HMAC: ...
elif sys.version_info >= (3, 4):
def new(key: _B, msg: Optional[_B] = ...,
digestmod: Optional[Union[str, Callable[[], _Hash], ModuleType]] = ...) -> HMAC: ...
else:
def new(key: _B, msg: Optional[_B] = ...,
digestmod: Optional[Union[Callable[[], _Hash], ModuleType]] = ...) -> HMAC: ...
class HMAC:
if sys.version_info >= (3,):
digest_size: int
if sys.version_info >= (3, 4):
block_size: int
name: str
def update(self, msg: _B) -> None: ...
def digest(self) -> bytes: ...
def hexdigest(self) -> str: ...
def copy(self) -> HMAC: ...
@overload
def compare_digest(a: bytearray, b: bytearray) -> bool: ...
@overload
def compare_digest(a: AnyStr, b: AnyStr) -> bool: ...
if sys.version_info >= (3, 7):
def digest(key: _B, msg: _B, digest: str) -> bytes: ...
@@ -0,0 +1,102 @@
import sys
from typing import (Optional, Sequence, Union, Generic, overload,
Iterable, Iterator, Sized, ContextManager, AnyStr)
ACCESS_DEFAULT: int
ACCESS_READ: int
ACCESS_WRITE: int
ACCESS_COPY: int
ALLOCATIONGRANULARITY: int
if sys.platform != 'win32':
MAP_ANON: int
MAP_ANONYMOUS: int
MAP_DENYWRITE: int
MAP_EXECUTABLE: int
MAP_PRIVATE: int
MAP_SHARED: int
PROT_EXEC: int
PROT_READ: int
PROT_WRITE: int
PAGESIZE: int
class _mmap(Generic[AnyStr]):
if sys.platform == 'win32':
def __init__(self, fileno: int, length: int,
tagname: Optional[str] = ..., access: int = ...,
offset: int = ...) -> None: ...
else:
def __init__(self,
fileno: int, length: int, flags: int = ...,
prot: int = ..., access: int = ...,
offset: int = ...) -> None: ...
def close(self) -> None: ...
def find(self, sub: AnyStr,
start: int = ..., end: int = ...) -> int: ...
if sys.version_info >= (3, 8):
def flush(self, offset: int = ..., size: int = ...) -> None: ...
else:
def flush(self, offset: int = ..., size: int = ...) -> int: ...
def move(self, dest: int, src: int, count: int) -> None: ...
def read(self, n: int = ...) -> AnyStr: ...
def read_byte(self) -> AnyStr: ...
def readline(self) -> AnyStr: ...
def resize(self, newsize: int) -> None: ...
def seek(self, pos: int, whence: int = ...) -> None: ...
def size(self) -> int: ...
def tell(self) -> int: ...
def write(self, bytes: AnyStr) -> None: ...
def write_byte(self, byte: AnyStr) -> None: ...
def __len__(self) -> int: ...
if sys.version_info >= (3,):
class mmap(_mmap[bytes], ContextManager[mmap], Iterable[bytes], Sized):
closed: bool
if sys.version_info >= (3, 8) and sys.platform != "win32":
def madvise(self, option: int, start: int = ..., length: int = ...) -> None: ...
def rfind(self, sub: bytes, start: int = ..., stop: int = ...) -> int: ...
@overload
def __getitem__(self, index: int) -> int: ...
@overload
def __getitem__(self, index: slice) -> bytes: ...
def __delitem__(self, index: Union[int, slice]) -> None: ...
@overload
def __setitem__(self, index: int, object: int) -> None: ...
@overload
def __setitem__(self, index: slice, object: bytes) -> None: ...
# Doesn't actually exist, but the object is actually iterable because it has __getitem__ and
# __len__, so we claim that there is also an __iter__ to help type checkers.
def __iter__(self) -> Iterator[bytes]: ...
else:
class mmap(_mmap[bytes], Sequence[bytes]):
def rfind(self, string: bytes, start: int = ..., stop: int = ...) -> int: ...
def __getitem__(self, index: Union[int, slice]) -> bytes: ...
def __getslice__(self, i: Optional[int], j: Optional[int]) -> bytes: ...
def __delitem__(self, index: Union[int, slice]) -> None: ...
def __setitem__(self, index: Union[int, slice], object: bytes) -> None: ...
if sys.version_info >= (3, 8):
MADV_NORMAL: int
MADV_RANDOM: int
MADV_SEQUENTIAL: int
MADV_WILLNEED: int
MADV_DONTNEED: int
MADV_REMOVE: int
MADV_DONTFORK: int
MADV_DOFORK: int
MADV_HWPOISON: int
MADV_MERGEABLE: int
MADV_UNMERGEABLE: int
MADV_SOFT_OFFLINE: int
MADV_HUGEPAGE: int
MADV_NOHUGEPAGE: int
MADV_DONTDUMP: int
MADV_DODUMP: int
MADV_FREE: int
MADV_NOSYNC: int
MADV_AUTOSYNC: int
MADV_NOCORE: int
MADV_CORE: int
MADV_PROTECT: int
@@ -0,0 +1,134 @@
import sys
from typing import List, Tuple, Union, Set, Optional, Dict, Container, Any, Type, Iterable, Sequence
from types import ModuleType
if sys.version_info >= (3, 8):
from typing import Literal
else:
from typing_extensions import Literal
if sys.platform == 'win32':
from _msi import _Database
AMD64: bool
if sys.version_info < (3, 7):
Itanium: bool
Win64: bool
datasizemask: Literal[0x00ff]
type_valid: Literal[0x0100]
type_localizable: Literal[0x0200]
typemask: Literal[0x0c00]
type_long: Literal[0x0000]
type_short: Literal[0x0400]
type_string: Literal[0x0c00]
type_binary: Literal[0x0800]
type_nullable: Literal[0x1000]
type_key: Literal[0x2000]
knownbits: Literal[0x3fff]
class Table:
name: str
fields: List[Tuple[int, str, int]]
def __init__(self, name: str) -> None: ...
def add_field(self, index: int, name: str, type: int) -> None: ...
def sql(self) -> str: ...
def create(self, db: _Database) -> None: ...
class _Unspecified: ...
def change_sequence(seq: Sequence[Tuple[str, Optional[str], int]], action: str, seqno: Union[int, Type[_Unspecified]] = ..., cond: Union[str, Type[_Unspecified]] = ...) -> None: ...
def add_data(db: _Database, table: str, values: Iterable[Tuple[Any, ...]]) -> None: ...
def add_stream(db: _Database, name: str, path: str) -> None: ...
def init_database(name: str, schema: ModuleType, ProductName: str, ProductCode: str, ProductVersion: str, Manufacturer: str) -> _Database: ...
def add_tables(db: _Database, module: ModuleType) -> None: ...
def make_id(str: str) -> str: ...
def gen_uuid() -> str: ...
class CAB:
name: str
files: List[Tuple[str, str]]
filenames: Set[str]
index: int
def __init__(self, name: str) -> None: ...
def gen_id(self, file: str) -> str: ...
def append(self, full: str, file: str, logical: str) -> Tuple[int, str]: ...
def commit(self, db: _Database) -> None: ...
_directories: Set[str]
class Directory:
db: _Database
cab: CAB
basedir: str
physical: str
logical: str
component: Optional[str]
short_names: Set[str]
ids: Set[str]
keyfiles: Dict[str, str]
componentflags: Optional[int]
absolute: str
def __init__(self, db: _Database, cab: CAB, basedir: str, physical: str, _logical: str, default: str, componentflags: Optional[int] = ...) -> None: ...
def start_component(self, component: Optional[str] = ..., feature: Optional[Feature] = ..., flags: Optional[int] = ..., keyfile: Optional[str] = ..., uuid: Optional[str] = ...) -> None: ...
def make_short(self, file: str) -> str: ...
def add_file(self, file: str, src: Optional[str] = ..., version: Optional[str] = ..., language: Optional[str] = ...) -> str: ...
def glob(self, pattern: str, exclude: Optional[Container[str]] = ...) -> List[str]: ...
def remove_pyc(self) -> None: ...
class Binary:
name: str
def __init__(self, fname: str) -> None: ...
def __repr__(self) -> str: ...
class Feature:
id: str
def __init__(self, db: _Database, id: str, title: str, desc: str, display: int, level: int = ..., parent: Optional[Feature] = ..., directory: Optional[str] = ..., attributes: int = ...) -> None: ...
def set_current(self) -> None: ...
class Control:
dlg: Dialog
name: str
def __init__(self, dlg: Dialog, name: str) -> None: ...
def event(self, event: str, argument: str, condition: str = ..., ordering: Optional[int] = ...) -> None: ...
def mapping(self, event: str, attribute: str) -> None: ...
def condition(self, action: str, condition: str) -> None: ...
class RadioButtonGroup(Control):
property: str
index: int
def __init__(self, dlg: Dialog, name: str, property: str) -> None: ...
def add(self, name: str, x: int, y: int, w: int, h: int, text: str, value: Optional[str] = ...) -> None: ...
class Dialog:
db: _Database
name: str
x: int
y: int
w: int
h: int
def __init__(self, db: _Database, name: str, x: int, y: int, w: int, h: int, attr: int, title: str, first: str, default: str, cancel: str) -> None: ...
def control(self, name: str, type: str, x: int, y: int, w: int, h: int, attr: int, prop: Optional[str], text: Optional[str], next: Optional[str], help: Optional[str]) -> Control: ...
def text(self, name: str, x: int, y: int, w: int, h: int, attr: int, text: Optional[str]) -> Control: ...
def bitmap(self, name: str, x: int, y: int, w: int, h: int, text: Optional[str]) -> Control: ...
def line(self, name: str, x: int, y: int, w: int, h: int) -> Control: ...
def pushbutton(self, name: str, x: int, y: int, w: int, h: int, attr: int, text: Optional[str], next: Optional[str]) -> Control: ...
def radiogroup(self, name: str, x: int, y: int, w: int, h: int, attr: int, prop: Optional[str], text: Optional[str], next: Optional[str]) -> RadioButtonGroup: ...
def checkbox(self, name: str, x: int, y: int, w: int, h: int, attr: int, prop: Optional[str], text: Optional[str], next: Optional[str]) -> Control: ...
@@ -0,0 +1,96 @@
import sys
from typing import List, Tuple, Optional
if sys.platform == 'win32':
from . import Table
_Validation: Table
ActionText: Table
AdminExecuteSequence: Table
Condition: Table
AdminUISequence: Table
AdvtExecuteSequence: Table
AdvtUISequence: Table
AppId: Table
AppSearch: Table
Property: Table
BBControl: Table
Billboard: Table
Feature: Table
Binary: Table
BindImage: Table
File: Table
CCPSearch: Table
CheckBox: Table
Class: Table
Component: Table
Icon: Table
ProgId: Table
ComboBox: Table
CompLocator: Table
Complus: Table
Directory: Table
Control: Table
Dialog: Table
ControlCondition: Table
ControlEvent: Table
CreateFolder: Table
CustomAction: Table
DrLocator: Table
DuplicateFile: Table
Environment: Table
Error: Table
EventMapping: Table
Extension: Table
MIME: Table
FeatureComponents: Table
FileSFPCatalog: Table
SFPCatalog: Table
Font: Table
IniFile: Table
IniLocator: Table
InstallExecuteSequence: Table
InstallUISequence: Table
IsolatedComponent: Table
LaunchCondition: Table
ListBox: Table
ListView: Table
LockPermissions: Table
Media: Table
MoveFile: Table
MsiAssembly: Table
MsiAssemblyName: Table
MsiDigitalCertificate: Table
MsiDigitalSignature: Table
MsiFileHash: Table
MsiPatchHeaders: Table
ODBCAttribute: Table
ODBCDriver: Table
ODBCDataSource: Table
ODBCSourceAttribute: Table
ODBCTranslator: Table
Patch: Table
PatchPackage: Table
PublishComponent: Table
RadioButton: Table
Registry: Table
RegLocator: Table
RemoveFile: Table
RemoveIniFile: Table
RemoveRegistry: Table
ReserveCost: Table
SelfReg: Table
ServiceControl: Table
ServiceInstall: Table
Shortcut: Table
Signature: Table
TextStyle: Table
TypeLib: Table
UIText: Table
Upgrade: Table
Verb: Table
tables: List[Table]
_Validation_records: List[Tuple[str, str, str, Optional[int], Optional[int], Optional[str], Optional[int], Optional[str], Optional[str], str]]
@@ -0,0 +1,15 @@
import sys
from typing import List, Tuple, Optional
if sys.platform == 'win32':
_SequenceType = List[Tuple[str, Optional[str], int]]
AdminExecuteSequence: _SequenceType
AdminUISequence: _SequenceType
AdvtExecuteSequence: _SequenceType
InstallExecuteSequence: _SequenceType
InstallUISequence: _SequenceType
tables: List[str]
@@ -0,0 +1,10 @@
import sys
from typing import List, Tuple, Optional
if sys.platform == 'win32':
ActionText: List[Tuple[str, str, Optional[str]]]
UIText: List[Tuple[str, Optional[str]]]
tables: List[str]
@@ -0,0 +1,87 @@
# Stubs for sunau (Python 2 and 3)
import sys
from typing import Any, NamedTuple, NoReturn, Optional, Text, IO, Union, Tuple
_File = Union[Text, IO[bytes]]
class Error(Exception): ...
AUDIO_FILE_MAGIC: int
AUDIO_FILE_ENCODING_MULAW_8: int
AUDIO_FILE_ENCODING_LINEAR_8: int
AUDIO_FILE_ENCODING_LINEAR_16: int
AUDIO_FILE_ENCODING_LINEAR_24: int
AUDIO_FILE_ENCODING_LINEAR_32: int
AUDIO_FILE_ENCODING_FLOAT: int
AUDIO_FILE_ENCODING_DOUBLE: int
AUDIO_FILE_ENCODING_ADPCM_G721: int
AUDIO_FILE_ENCODING_ADPCM_G722: int
AUDIO_FILE_ENCODING_ADPCM_G723_3: int
AUDIO_FILE_ENCODING_ADPCM_G723_5: int
AUDIO_FILE_ENCODING_ALAW_8: int
AUDIO_UNKNOWN_SIZE: int
if sys.version_info < (3, 0):
_sunau_params = Tuple[int, int, int, int, str, str]
else:
class _sunau_params(NamedTuple):
nchannels: int
sampwidth: int
framerate: int
nframes: int
comptype: str
compname: str
class Au_read:
def __init__(self, f: _File) -> None: ...
if sys.version_info >= (3, 3):
def __enter__(self) -> Au_read: ...
def __exit__(self, *args: Any) -> None: ...
def getfp(self) -> Optional[IO[bytes]]: ...
def rewind(self) -> None: ...
def close(self) -> None: ...
def tell(self) -> int: ...
def getnchannels(self) -> int: ...
def getnframes(self) -> int: ...
def getsampwidth(self) -> int: ...
def getframerate(self) -> int: ...
def getcomptype(self) -> str: ...
def getcompname(self) -> str: ...
def getparams(self) -> _sunau_params: ...
def getmarkers(self) -> None: ...
def getmark(self, id: Any) -> NoReturn: ...
def setpos(self, pos: int) -> None: ...
def readframes(self, nframes: int) -> Optional[bytes]: ...
class Au_write:
def __init__(self, f: _File) -> None: ...
if sys.version_info >= (3, 3):
def __enter__(self) -> Au_write: ...
def __exit__(self, *args: Any) -> None: ...
def setnchannels(self, nchannels: int) -> None: ...
def getnchannels(self) -> int: ...
def setsampwidth(self, sampwidth: int) -> None: ...
def getsampwidth(self) -> int: ...
def setframerate(self, framerate: float) -> None: ...
def getframerate(self) -> int: ...
def setnframes(self, nframes: int) -> None: ...
def getnframes(self) -> int: ...
def setcomptype(self, comptype: str, compname: str) -> None: ...
def getcomptype(self) -> str: ...
def getcompname(self) -> str: ...
def setparams(self, params: _sunau_params) -> None: ...
def getparams(self) -> _sunau_params: ...
def setmark(self, id: Any, pos: Any, name: Any) -> NoReturn: ...
def getmark(self, id: Any) -> NoReturn: ...
def getmarkers(self) -> None: ...
def tell(self) -> int: ...
# should be any bytes-like object after 3.4, but we don't have a type for that
def writeframesraw(self, data: bytes) -> None: ...
def writeframes(self, data: bytes) -> None: ...
def close(self) -> None: ...
# Returns a Au_read if mode is rb and Au_write if mode is wb
def open(f: _File, mode: Optional[str] = ...) -> Any: ...
if sys.version_info < (3, 9):
openfp = open
@@ -0,0 +1,76 @@
# Stubs for wave (Python 2 and 3)
import sys
from typing import (
Any, NamedTuple, NoReturn, Optional, Text, BinaryIO, Union, Tuple, IO
)
_File = Union[Text, IO[bytes]]
class Error(Exception): ...
WAVE_FORMAT_PCM: int
if sys.version_info < (3, 0):
_wave_params = Tuple[int, int, int, int, str, str]
else:
class _wave_params(NamedTuple):
nchannels: int
sampwidth: int
framerate: int
nframes: int
comptype: str
compname: str
class Wave_read:
def __init__(self, f: _File) -> None: ...
if sys.version_info >= (3, 0):
def __enter__(self) -> Wave_read: ...
def __exit__(self, *args: Any) -> None: ...
def getfp(self) -> Optional[BinaryIO]: ...
def rewind(self) -> None: ...
def close(self) -> None: ...
def tell(self) -> int: ...
def getnchannels(self) -> int: ...
def getnframes(self) -> int: ...
def getsampwidth(self) -> int: ...
def getframerate(self) -> int: ...
def getcomptype(self) -> str: ...
def getcompname(self) -> str: ...
def getparams(self) -> _wave_params: ...
def getmarkers(self) -> None: ...
def getmark(self, id: Any) -> NoReturn: ...
def setpos(self, pos: int) -> None: ...
def readframes(self, nframes: int) -> bytes: ...
class Wave_write:
def __init__(self, f: _File) -> None: ...
if sys.version_info >= (3, 0):
def __enter__(self) -> Wave_write: ...
def __exit__(self, *args: Any) -> None: ...
def setnchannels(self, nchannels: int) -> None: ...
def getnchannels(self) -> int: ...
def setsampwidth(self, sampwidth: int) -> None: ...
def getsampwidth(self) -> int: ...
def setframerate(self, framerate: float) -> None: ...
def getframerate(self) -> int: ...
def setnframes(self, nframes: int) -> None: ...
def getnframes(self) -> int: ...
def setcomptype(self, comptype: str, compname: str) -> None: ...
def getcomptype(self) -> str: ...
def getcompname(self) -> str: ...
def setparams(self, params: _wave_params) -> None: ...
def getparams(self) -> _wave_params: ...
def setmark(self, id: Any, pos: Any, name: Any) -> NoReturn: ...
def getmark(self, id: Any) -> NoReturn: ...
def getmarkers(self) -> None: ...
def tell(self) -> int: ...
# should be any bytes-like object after 3.4, but we don't have a type for that
def writeframesraw(self, data: bytes) -> None: ...
def writeframes(self, data: bytes) -> None: ...
def close(self) -> None: ...
# Returns a Wave_read if mode is rb and Wave_write if mode is wb
def open(f: _File, mode: Optional[str] = ...) -> Any: ...
if sys.version_info < (3, 9):
openfp = open
@@ -0,0 +1,87 @@
import sys
from abc import abstractmethod
from types import TracebackType
from typing import Optional, Dict, MutableMapping, Type, Text, Callable, List, Tuple, IO
from .headers import Headers
from .types import WSGIApplication, WSGIEnvironment, StartResponse, InputStream, ErrorStream
from .util import FileWrapper, guess_scheme
_exc_info = Tuple[Optional[Type[BaseException]],
Optional[BaseException],
Optional[TracebackType]]
def format_date_time(timestamp: Optional[float]) -> str: ... # undocumented
if sys.version_info >= (3, 2):
def read_environ() -> Dict[str, str]: ...
class BaseHandler:
wsgi_version: Tuple[int, int] # undocumented
wsgi_multithread: bool
wsgi_multiprocess: bool
wsgi_run_once: bool
origin_server: bool
http_version: str
server_software: Optional[str]
os_environ: MutableMapping[str, str]
wsgi_file_wrapper: Optional[Type[FileWrapper]]
headers_class: Type[Headers] # undocumented
traceback_limit: Optional[int]
error_status: str
error_headers: List[Tuple[Text, Text]]
error_body: bytes
def run(self, application: WSGIApplication) -> None: ...
def setup_environ(self) -> None: ...
def finish_response(self) -> None: ...
def get_scheme(self) -> str: ...
def set_content_length(self) -> None: ...
def cleanup_headers(self) -> None: ...
def start_response(self, status: Text, headers: List[Tuple[Text, Text]], exc_info: Optional[_exc_info] = ...) -> Callable[[bytes], None]: ...
def send_preamble(self) -> None: ...
def write(self, data: bytes) -> None: ...
def sendfile(self) -> bool: ...
def finish_content(self) -> None: ...
def close(self) -> None: ...
def send_headers(self) -> None: ...
def result_is_file(self) -> bool: ...
def client_is_modern(self) -> bool: ...
def log_exception(self, exc_info: _exc_info) -> None: ...
def handle_error(self) -> None: ...
def error_output(self, environ: WSGIEnvironment, start_response: StartResponse) -> List[bytes]: ...
@abstractmethod
def _write(self, data: bytes) -> None: ...
@abstractmethod
def _flush(self) -> None: ...
@abstractmethod
def get_stdin(self) -> InputStream: ...
@abstractmethod
def get_stderr(self) -> ErrorStream: ...
@abstractmethod
def add_cgi_vars(self) -> None: ...
class SimpleHandler(BaseHandler):
stdin: InputStream
stdout: IO[bytes]
stderr: ErrorStream
base_env: MutableMapping[str, str]
def __init__(self, stdin: InputStream, stdout: IO[bytes], stderr: ErrorStream, environ: MutableMapping[str, str], multithread: bool = ..., multiprocess: bool = ...) -> None: ...
def get_stdin(self) -> InputStream: ...
def get_stderr(self) -> ErrorStream: ...
def add_cgi_vars(self) -> None: ...
def _write(self, data: bytes) -> None: ...
def _flush(self) -> None: ...
class BaseCGIHandler(SimpleHandler): ...
class CGIHandler(BaseCGIHandler):
def __init__(self) -> None: ...
class IISCGIHandler(BaseCGIHandler):
def __init__(self) -> None: ...
@@ -0,0 +1,31 @@
import sys
from typing import overload, Pattern, Optional, List, Tuple
_HeaderList = List[Tuple[str, str]]
tspecials: Pattern[str] # undocumented
class Headers:
if sys.version_info < (3, 5):
def __init__(self, headers: _HeaderList) -> None: ...
else:
def __init__(self, headers: Optional[_HeaderList] = ...) -> None: ...
def __len__(self) -> int: ...
def __setitem__(self, name: str, val: str) -> None: ...
def __delitem__(self, name: str) -> None: ...
def __getitem__(self, name: str) -> Optional[str]: ...
if sys.version_info < (3,):
def has_key(self, name: str) -> bool: ...
def __contains__(self, name: str) -> bool: ...
def get_all(self, name: str) -> List[str]: ...
@overload
def get(self, name: str, default: str) -> str: ...
@overload
def get(self, name: str, default: Optional[str] = ...) -> Optional[str]: ...
def keys(self) -> List[str]: ...
def values(self) -> List[str]: ...
def items(self) -> _HeaderList: ...
if sys.version_info >= (3,):
def __bytes__(self) -> bytes: ...
def setdefault(self, name: str, value: str) -> str: ...
def add_header(self, _name: str, _value: Optional[str], **_params: Optional[str]) -> None: ...
@@ -0,0 +1,40 @@
import sys
from typing import Optional, List, Type, TypeVar, overload
from .handlers import SimpleHandler
from .types import WSGIApplication, WSGIEnvironment, StartResponse, ErrorStream
if sys.version_info < (3,):
from BaseHTTPServer import BaseHTTPRequestHandler, HTTPServer
else:
from http.server import HTTPServer, BaseHTTPRequestHandler
server_version: str # undocumented
sys_version: str # undocumented
software_version: str # undocumented
class ServerHandler(SimpleHandler): # undocumented
server_software: str
def close(self) -> None: ...
class WSGIServer(HTTPServer):
application: Optional[WSGIApplication]
base_environ: WSGIEnvironment # only available after call to setup_environ()
def setup_environ(self) -> None: ...
def get_app(self) -> Optional[WSGIApplication]: ...
def set_app(self, application: Optional[WSGIApplication]) -> None: ...
class WSGIRequestHandler(BaseHTTPRequestHandler):
server_version: str
def get_environ(self) -> WSGIEnvironment: ...
def get_stderr(self) -> ErrorStream: ...
def handle(self) -> None: ...
def demo_app(environ: WSGIEnvironment, start_response: StartResponse) -> List[bytes]: ...
_S = TypeVar("_S", bound=WSGIServer)
@overload
def make_server(host: str, port: int, app: WSGIApplication, *, handler_class: Type[WSGIRequestHandler] = ...) -> WSGIServer: ...
@overload
def make_server(host: str, port: int, app: WSGIApplication, server_class: Type[_S], handler_class: Type[WSGIRequestHandler] = ...) -> _S: ...
@@ -0,0 +1,44 @@
# Type declaration for a WSGI Function
#
# wsgiref/types.py doesn't exist and neither do the types defined in this
# file. They are provided for type checking purposes.
#
# This means you cannot simply import wsgiref.types in your code. Instead,
# use the `TYPE_CHECKING` flag from the typing module:
#
# from typing import TYPE_CHECKING
#
# if TYPE_CHECKING:
# from wsgiref.types import WSGIApplication
#
# This import is now only taken into account by the type checker. Consequently,
# you need to use 'WSGIApplication' and not simply WSGIApplication when type
# hinting your code. Otherwise Python will raise NameErrors.
from sys import _OptExcInfo
from typing import Callable, Dict, Iterable, List, Any, Text, Protocol, Tuple, Optional
class StartResponse(Protocol):
def __call__(self, status: str, headers: List[Tuple[str, str]], exc_info: Optional[_OptExcInfo] = ...) -> Callable[[bytes], Any]: ...
WSGIEnvironment = Dict[Text, Any]
WSGIApplication = Callable[[WSGIEnvironment, StartResponse], Iterable[bytes]]
# WSGI input streams per PEP 3333
class InputStream(Protocol):
def read(self, size: int = ...) -> bytes: ...
def readline(self, size: int = ...) -> bytes: ...
def readlines(self, hint: int = ...) -> List[bytes]: ...
def __iter__(self) -> Iterable[bytes]: ...
# WSGI error streams per PEP 3333
class ErrorStream(Protocol):
def flush(self) -> None: ...
def write(self, s: str) -> None: ...
def writelines(self, seq: List[str]) -> None: ...
class _Readable(Protocol):
def read(self, size: int = ...) -> bytes: ...
# Optional file wrapper in wsgi.file_wrapper
class FileWrapper(Protocol):
def __call__(self, file: _Readable, block_size: int = ...) -> Iterable[bytes]: ...
@@ -0,0 +1,23 @@
import sys
from typing import IO, Any, Callable, Optional
from .types import WSGIEnvironment
class FileWrapper:
filelike: IO[bytes]
blksize: int
close: Callable[[], None] # only exists if filelike.close exists
def __init__(self, filelike: IO[bytes], blksize: int = ...) -> None: ...
def __getitem__(self, key: Any) -> bytes: ...
def __iter__(self) -> FileWrapper: ...
if sys.version_info < (3,):
def next(self) -> bytes: ...
else:
def __next__(self) -> bytes: ...
def guess_scheme(environ: WSGIEnvironment) -> str: ...
def application_uri(environ: WSGIEnvironment) -> str: ...
def request_uri(environ: WSGIEnvironment, include_query: bool = ...) -> str: ...
def shift_path_info(environ: WSGIEnvironment) -> Optional[str]: ...
def setup_testing_defaults(environ: WSGIEnvironment) -> None: ...
def is_hop_by_hop(header_name: str) -> bool: ...
@@ -0,0 +1,53 @@
import sys
from typing import Any, Iterable, Iterator, Optional, NoReturn, Callable
from wsgiref.types import WSGIApplication, InputStream, ErrorStream
class WSGIWarning(Warning): ...
def validator(application: WSGIApplication) -> WSGIApplication: ...
class InputWrapper:
input: InputStream
def __init__(self, wsgi_input: InputStream) -> None: ...
if sys.version_info < (3,):
def read(self, size: int = ...) -> bytes: ...
def readline(self) -> bytes: ...
else:
def read(self, size: int) -> bytes: ...
def readline(self, size: int = ...) -> bytes: ...
def readlines(self, hint: int = ...) -> bytes: ...
def __iter__(self) -> Iterable[bytes]: ...
def close(self) -> NoReturn: ...
class ErrorWrapper:
errors: ErrorStream
def __init__(self, wsgi_errors: ErrorStream) -> None: ...
def write(self, s: str) -> None: ...
def flush(self) -> None: ...
def writelines(self, seq: Iterable[str]) -> None: ...
def close(self) -> NoReturn: ...
class WriteWrapper:
writer: Callable[[bytes], Any]
def __init__(self, wsgi_writer: Callable[[bytes], Any]) -> None: ...
def __call__(self, s: bytes) -> None: ...
class PartialIteratorWrapper:
iterator: Iterator[bytes]
def __init__(self, wsgi_iterator: Iterator[bytes]) -> None: ...
def __iter__(self) -> IteratorWrapper: ...
class IteratorWrapper:
original_iterator: Iterator[bytes]
iterator: Iterator[bytes]
closed: bool
check_start_response: Optional[bool]
def __init__(self, wsgi_iterator: Iterator[bytes], check_start_response: Optional[bool]) -> None: ...
def __iter__(self) -> IteratorWrapper: ...
if sys.version_info < (3,):
def next(self) -> bytes: ...
else:
def __next__(self) -> bytes: ...
def close(self) -> None: ...
def __del__(self) -> None: ...
@@ -0,0 +1,11 @@
from typing import Type, TypeVar, Tuple, Any, Dict, NewType
_T = TypeVar('_T')
# TODO: Change the return into a NewType bound to int after pytype/#597
def get_cache_token() -> object: ...
class ABCMeta(type):
def __new__(mcls, __name: str, __bases: Tuple[Type[Any], ...], __namespace: Dict[str, Any]) -> ABCMeta: ...
def register(cls, subclass: Type[_T]) -> Type[_T]: ...
@@ -0,0 +1,2 @@
def getpreferredencoding(do_setlocale: bool = ...) -> str: ...
@@ -0,0 +1,14 @@
from typing import Dict, Tuple
import sys
IMPORT_MAPPING: Dict[str, str]
NAME_MAPPING: Dict[Tuple[str, str], Tuple[str, str]]
PYTHON2_EXCEPTIONS: Tuple[str, ...]
MULTIPROCESSING_EXCEPTIONS: Tuple[str, ...]
REVERSE_IMPORT_MAPPING: Dict[str, str]
REVERSE_NAME_MAPPING: Dict[Tuple[str, str], Tuple[str, str]]
PYTHON3_OSERROR_EXCEPTIONS: Tuple[str, ...]
if sys.version_info >= (3, 6):
PYTHON3_IMPORTERROR_EXCEPTIONS: Tuple[str, ...]
@@ -0,0 +1,24 @@
import sys
from typing import Optional, NoReturn, ClassVar, Iterable
if sys.version_info >= (3, 8):
from typing import Literal
else:
from typing_extensions import Literal
class Quitter:
name: str
eof: str
def __init__(self, name: str, eof: str) -> None: ...
def __call__(self, code: Optional[int] = ...) -> NoReturn: ...
class _Printer:
MAXLINES: ClassVar[Literal[23]]
def __init__(self, name: str, data: str, files: Iterable[str] = ..., dirs: Iterable[str] = ...) -> None: ...
def __call__(self) -> None: ...
class _Helper:
def __call__(self, request: object) -> None: ...
@@ -0,0 +1,12 @@
from typing import Optional, Sequence, Tuple
from tracemalloc import _TraceTupleT, _FrameTupleT
def _get_object_traceback(__obj) -> Optional[Sequence[_FrameTupleT]]: ...
def _get_traces() -> Sequence[_TraceTupleT]: ...
def clear_traces() -> None: ...
def get_traceback_limit() -> int: ...
def get_traced_memory() -> Tuple[int, int]: ...
def get_tracemalloc_memory() -> int: ...
def is_tracing() -> bool: ...
def start(__nframe: int = ...) -> None: ...
def stop() -> None: ...
@@ -0,0 +1,107 @@
import sys
from typing import Any, Union, Tuple, Optional, overload, Dict, NoReturn, Sequence
if sys.version_info >= (3, 8):
from typing import Literal
else:
from typing_extensions import Literal
CREATE_NEW_CONSOLE: int
CREATE_NEW_PROCESS_GROUP: int
DUPLICATE_CLOSE_SOURCE: int
DUPLICATE_SAME_ACCESS: int
ERROR_ALREADY_EXISTS: int
ERROR_BROKEN_PIPE: int
ERROR_IO_PENDING: int
ERROR_MORE_DATA: int
ERROR_NETNAME_DELETED: int
ERROR_NO_DATA: int
ERROR_NO_SYSTEM_RESOURCES: int
ERROR_OPERATION_ABORTED: int
ERROR_PIPE_BUSY: int
ERROR_PIPE_CONNECTED: int
ERROR_SEM_TIMEOUT: int
FILE_FLAG_FIRST_PIPE_INSTANCE: int
FILE_FLAG_OVERLAPPED: int
FILE_GENERIC_READ: int
FILE_GENERIC_WRITE: int
GENERIC_READ: int
GENERIC_WRITE: int
INFINITE: int
NMPWAIT_WAIT_FOREVER: int
NULL: int
OPEN_EXISTING: int
PIPE_ACCESS_DUPLEX: int
PIPE_ACCESS_INBOUND: int
PIPE_READMODE_MESSAGE: int
PIPE_TYPE_MESSAGE: int
PIPE_UNLIMITED_INSTANCES: int
PIPE_WAIT: int
PROCESS_ALL_ACCESS: int
PROCESS_DUP_HANDLE: int
STARTF_USESHOWWINDOW: int
STARTF_USESTDHANDLES: int
STD_ERROR_HANDLE: int
STD_INPUT_HANDLE: int
STD_OUTPUT_HANDLE: int
STILL_ACTIVE: int
SW_HIDE: int
WAIT_ABANDONED_0: int
WAIT_OBJECT_0: int
WAIT_TIMEOUT: int
def CloseHandle(__handle: int) -> None: ...
@overload
def ConnectNamedPipe(handle: int, overlapped: Literal[True]) -> Overlapped: ...
@overload
def ConnectNamedPipe(handle: int, overlapped: Literal[False] = ...) -> None: ...
@overload
def ConnectNamedPipe(handle: int, overlapped: bool) -> Optional[Overlapped]: ...
def CreateFile(__file_name: str, __desired_access: int, __share_mode: int, __security_attributes: int, __creation_disposition: int, __flags_and_attributes: int, __template_file: int) -> int: ...
def CreateJunction(__src_path: str, __dst_path: str) -> None: ...
def CreateNamedPipe(__name: str, __open_mode: int, __pipe_mode: int, __max_instances: int, __out_buffer_size: int, __in_buffer_size: int, __default_timeout: int, __security_attributes: int) -> int: ...
def CreatePipe(__pipe_attrs: Any, __size: int) -> Tuple[int, int]: ...
def CreateProcess(__application_name: Optional[str], __command_line: Optional[str], __proc_attrs: Any, __thread_attrs: Any, __inherit_handles: bool, __creation_flags: int, __env_mapping: Dict[str, str], __current_directory: Optional[str], __startup_info: Any) -> Tuple[int, int, int, int]: ...
def DuplicateHandle(__source_process_handle: int, __source_handle: int, __target_process_handle: int, __desired_access: int, __inherit_handle: bool, __options: int = ...) -> int: ...
def ExitProcess(__ExitCode: int) -> NoReturn: ...
if sys.version_info >= (3, 7):
def GetACP() -> int: ...
def GetFileType(handle: int) -> int: ...
def GetCurrentProcess() -> int: ...
def GetExitCodeProcess(__process: int) -> int: ...
def GetLastError() -> int: ...
def GetModuleFileName(__module_handle: int) -> str: ...
def GetStdHandle(__std_handle: int) -> int: ...
def GetVersion() -> int: ...
def OpenProcess(__desired_access: int, __inherit_handle: bool, __process_id: int) -> int: ...
def PeekNamedPipe(__handle: int, __size: int = ...) -> Union[Tuple[int, int], Tuple[bytes, int, int]]: ...
@overload
def ReadFile(handle: int, size: int, overlapped: Literal[True]) -> Tuple[Overlapped, int]: ...
@overload
def ReadFile(handle: int, size: int, overlapped: Literal[False] = ...) -> Tuple[bytes, int]: ...
@overload
def ReadFile(handle: int, size: int, overlapped: Union[int, bool]) -> Tuple[Any, int]: ...
def SetNamedPipeHandleState(__named_pipe: int, __mode: Optional[int], __max_collection_count: Optional[int], __collect_data_timeout: Optional[int]) -> None: ...
def TerminateProcess(__handle: int, __exit_code: int) -> None: ...
def WaitForMultipleObjects(__handle_seq: Sequence[int], __wait_flag: bool, __milliseconds: int = ...) -> int: ...
def WaitForSingleObject(__handle: int, __milliseconds: int) -> int: ...
def WaitNamedPipe(__name: str, __timeout: int) -> None: ...
@overload
def WriteFile(handle: int, buffer: bytes, overlapped: Literal[True]) -> Tuple[Overlapped, int]: ...
@overload
def WriteFile(handle: int, buffer: bytes, overlapped: Literal[False] = ...) -> Tuple[int, int]: ...
@overload
def WriteFile(handle: int, buffer: bytes, overlapped: Union[int, bool]) -> Tuple[Any, int]: ...
class Overlapped:
event: int = ...
def GetOverlappedResult(self, __wait: bool) -> Tuple[int, int]: ...
def cancel(self) -> None: ...
def getbuffer(self) -> Optional[bytes]: ...
@@ -0,0 +1,113 @@
# Stubs for compileall (Python 3)
import os
import sys
from typing import Any, Optional, Union, Pattern
if sys.version_info < (3, 6):
_Path = Union[str, bytes]
_SuccessType = bool
else:
_Path = Union[str, bytes, os.PathLike]
_SuccessType = int
if sys.version_info >= (3, 7):
from py_compile import PycInvalidationMode
if sys.version_info >= (3, 9):
def compile_dir(
dir: _Path,
maxlevels: Optional[int] = ...,
ddir: Optional[_Path] = ...,
force: bool = ...,
rx: Optional[Pattern[Any]] = ...,
quiet: int = ...,
legacy: bool = ...,
optimize: int = ...,
workers: int = ...,
invalidation_mode: Optional[PycInvalidationMode] = ...,
*,
stripdir: Optional[str] = ..., # TODO: change to Optional[_Path] once https://bugs.python.org/issue40447 is resolved
prependdir: Optional[_Path] = ...,
limit_sl_dest: Optional[_Path] = ...,
) -> _SuccessType: ...
def compile_file(
fullname: _Path,
ddir: Optional[_Path] = ...,
force: bool = ...,
rx: Optional[Pattern[Any]] = ...,
quiet: int = ...,
legacy: bool = ...,
optimize: int = ...,
invalidation_mode: Optional[PycInvalidationMode] = ...,
*,
stripdir: Optional[str] = ..., # TODO: change to Optional[_Path] once https://bugs.python.org/issue40447 is resolved
prependdir: Optional[_Path] = ...,
limit_sl_dest: Optional[_Path] = ...,
) -> _SuccessType: ...
elif sys.version_info >= (3, 7):
def compile_dir(
dir: _Path,
maxlevels: int = ...,
ddir: Optional[_Path] = ...,
force: bool = ...,
rx: Optional[Pattern[Any]] = ...,
quiet: int = ...,
legacy: bool = ...,
optimize: int = ...,
workers: int = ...,
invalidation_mode: Optional[PycInvalidationMode] = ...,
) -> _SuccessType: ...
def compile_file(
fullname: _Path,
ddir: Optional[_Path] = ...,
force: bool = ...,
rx: Optional[Pattern[Any]] = ...,
quiet: int = ...,
legacy: bool = ...,
optimize: int = ...,
invalidation_mode: Optional[PycInvalidationMode] = ...,
) -> _SuccessType: ...
else:
# rx can be any object with a 'search' method; once we have Protocols we can change the type
def compile_dir(
dir: _Path,
maxlevels: int = ...,
ddir: Optional[_Path] = ...,
force: bool = ...,
rx: Optional[Pattern[Any]] = ...,
quiet: int = ...,
legacy: bool = ...,
optimize: int = ...,
workers: int = ...,
) -> _SuccessType: ...
def compile_file(
fullname: _Path,
ddir: Optional[_Path] = ...,
force: bool = ...,
rx: Optional[Pattern[Any]] = ...,
quiet: int = ...,
legacy: bool = ...,
optimize: int = ...,
) -> _SuccessType: ...
if sys.version_info >= (3, 7):
def compile_path(
skip_curdir: bool = ...,
maxlevels: int = ...,
force: bool = ...,
quiet: int = ...,
legacy: bool = ...,
optimize: int = ...,
invalidation_mode: Optional[PycInvalidationMode] = ...,
) -> _SuccessType: ...
else:
def compile_path(
skip_curdir: bool = ...,
maxlevels: int = ...,
force: bool = ...,
quiet: int = ...,
legacy: bool = ...,
optimize: int = ...,
) -> _SuccessType: ...
+169
View File
@@ -0,0 +1,169 @@
import io
import sys
from os.path import _PathType
from typing import IO, Any, Mapping, Optional, Sequence, TextIO, TypeVar, Union, overload
if sys.version_info >= (3, 8):
from typing import Literal
else:
from typing_extensions import Literal
_OpenBinaryWritingMode = Literal["w", "wb", "x", "xb", "a", "ab"]
_OpenTextWritingMode = Literal["wt", "xt", "at"]
_PathOrFile = Union[_PathType, IO[bytes]]
_FilterChain = Sequence[Mapping[str, Any]]
_T = TypeVar("_T")
FORMAT_AUTO: int
FORMAT_XZ: int
FORMAT_ALONE: int
FORMAT_RAW: int
CHECK_NONE: int
CHECK_CRC32: int
CHECK_CRC64: int
CHECK_SHA256: int
CHECK_ID_MAX: int
CHECK_UNKNOWN: int
FILTER_LZMA1: int
FILTER_LZMA2: int
FILTER_DELTA: int
FILTER_X86: int
FILTER_IA64: int
FILTER_ARM: int
FILTER_ARMTHUMB: int
FILTER_SPARC: int
FILTER_POWERPC: int
MF_HC3: int
MF_HC4: int
MF_BT2: int
MF_BT3: int
MF_BT4: int
MODE_FAST: int
MODE_NORMAL: int
PRESET_DEFAULT: int
PRESET_EXTREME: int
# from _lzma.c
class LZMADecompressor(object):
def __init__(
self, format: Optional[int] = ..., memlimit: Optional[int] = ..., filters: Optional[_FilterChain] = ...
) -> None: ...
def decompress(self, data: bytes, max_length: int = ...) -> bytes: ...
@property
def check(self) -> int: ...
@property
def eof(self) -> bool: ...
@property
def unused_data(self) -> bytes: ...
@property
def needs_input(self) -> bool: ...
# from _lzma.c
class LZMACompressor(object):
def __init__(
self, format: Optional[int] = ..., check: int = ..., preset: Optional[int] = ..., filters: Optional[_FilterChain] = ...
) -> None: ...
def compress(self, data: bytes) -> bytes: ...
def flush(self) -> bytes: ...
class LZMAError(Exception): ...
class LZMAFile(io.BufferedIOBase, IO[bytes]):
def __init__(
self,
filename: Optional[_PathOrFile] = ...,
mode: str = ...,
*,
format: Optional[int] = ...,
check: int = ...,
preset: Optional[int] = ...,
filters: Optional[_FilterChain] = ...,
) -> None: ...
def __enter__(self: _T) -> _T: ...
def close(self) -> None: ...
@property
def closed(self) -> bool: ...
def fileno(self) -> int: ...
def seekable(self) -> bool: ...
def readable(self) -> bool: ...
def writable(self) -> bool: ...
def peek(self, size: int = ...) -> bytes: ...
def read(self, size: Optional[int] = ...) -> bytes: ...
def read1(self, size: int = ...) -> bytes: ...
def readline(self, size: Optional[int] = ...) -> bytes: ...
def write(self, data: bytes) -> int: ...
def seek(self, offset: int, whence: int = ...) -> int: ...
def tell(self) -> int: ...
@overload
def open(
filename: _PathOrFile,
mode: Literal["r", "rb"] = ...,
*,
format: Optional[int] = ...,
check: Literal[-1] = ...,
preset: None = ...,
filters: Optional[_FilterChain] = ...,
encoding: None = ...,
errors: None = ...,
newline: None = ...,
) -> LZMAFile: ...
@overload
def open(
filename: _PathOrFile,
mode: _OpenBinaryWritingMode,
*,
format: Optional[int] = ...,
check: int = ...,
preset: Optional[int] = ...,
filters: Optional[_FilterChain] = ...,
encoding: None = ...,
errors: None = ...,
newline: None = ...,
) -> LZMAFile: ...
@overload
def open(
filename: _PathType,
mode: Literal["rt"],
*,
format: Optional[int] = ...,
check: Literal[-1] = ...,
preset: None = ...,
filters: Optional[_FilterChain] = ...,
encoding: Optional[str] = ...,
errors: Optional[str] = ...,
newline: Optional[str] = ...,
) -> TextIO: ...
@overload
def open(
filename: _PathType,
mode: _OpenTextWritingMode,
*,
format: Optional[int] = ...,
check: int = ...,
preset: Optional[int] = ...,
filters: Optional[_FilterChain] = ...,
encoding: Optional[str] = ...,
errors: Optional[str] = ...,
newline: Optional[str] = ...,
) -> TextIO: ...
@overload
def open(
filename: _PathOrFile,
mode: str,
*,
format: Optional[int] = ...,
check: int = ...,
preset: Optional[int] = ...,
filters: Optional[_FilterChain] = ...,
encoding: Optional[str] = ...,
errors: Optional[str] = ...,
newline: Optional[str] = ...,
) -> Union[LZMAFile, TextIO]: ...
def compress(
data: bytes, format: int = ..., check: int = ..., preset: Optional[int] = ..., filters: Optional[_FilterChain] = ...
) -> bytes: ...
def decompress(data: bytes, format: int = ..., memlimit: Optional[int] = ..., filters: Optional[_FilterChain] = ...) -> bytes: ...
def is_check_supported(check: int) -> bool: ...
@@ -0,0 +1,79 @@
# Stubs for random
# Ron Murawski <ron@horizonchess.com>
# Updated by Jukka Lehtosalo
# based on http://docs.python.org/3.2/library/random.html
# ----- random classes -----
import _random
import sys
from typing import Any, TypeVar, Sequence, Iterable, List, Callable, AbstractSet, Union, Optional, Tuple
_T = TypeVar('_T')
class Random(_random.Random):
def __init__(self, x: Any = ...) -> None: ...
def seed(self, a: Any = ..., version: int = ...) -> None: ...
def getstate(self) -> Tuple[Any, ...]: ...
def setstate(self, state: Tuple[Any, ...]) -> None: ...
def getrandbits(self, __k: int) -> int: ...
def randrange(self, start: int, stop: Union[int, None] = ..., step: int = ...) -> int: ...
def randint(self, a: int, b: int) -> int: ...
if sys.version_info >= (3, 9):
def randbytes(self, n: int) -> bytes: ...
def choice(self, seq: Sequence[_T]) -> _T: ...
if sys.version_info >= (3, 6):
def choices(self, population: Sequence[_T], weights: Optional[Sequence[float]] = ..., *, cum_weights: Optional[Sequence[float]] = ..., k: int = ...) -> List[_T]: ...
def shuffle(self, x: List[Any], random: Union[Callable[[], float], None] = ...) -> None: ...
if sys.version_info >= (3, 9):
def sample(self, population: Union[Sequence[_T], AbstractSet[_T]], k: int, *, counts: Optional[Iterable[_T]] = ...) -> List[_T]: ...
else:
def sample(self, population: Union[Sequence[_T], AbstractSet[_T]], k: int) -> List[_T]: ...
def random(self) -> float: ...
def uniform(self, a: float, b: float) -> float: ...
def triangular(self, low: float = ..., high: float = ..., mode: Optional[float] = ...) -> float: ...
def betavariate(self, alpha: float, beta: float) -> float: ...
def expovariate(self, lambd: float) -> float: ...
def gammavariate(self, alpha: float, beta: float) -> float: ...
def gauss(self, mu: float, sigma: float) -> float: ...
def lognormvariate(self, mu: float, sigma: float) -> float: ...
def normalvariate(self, mu: float, sigma: float) -> float: ...
def vonmisesvariate(self, mu: float, kappa: float) -> float: ...
def paretovariate(self, alpha: float) -> float: ...
def weibullvariate(self, alpha: float, beta: float) -> float: ...
# SystemRandom is not implemented for all OS's; good on Windows & Linux
class SystemRandom(Random):
...
# ----- random function stubs -----
def seed(a: Any = ..., version: int = ...) -> None: ...
def getstate() -> object: ...
def setstate(state: object) -> None: ...
def getrandbits(__k: int) -> int: ...
def randrange(start: int, stop: Union[None, int] = ..., step: int = ...) -> int: ...
def randint(a: int, b: int) -> int: ...
if sys.version_info >= (3, 9):
def randbytes(n: int) -> bytes: ...
def choice(seq: Sequence[_T]) -> _T: ...
if sys.version_info >= (3, 6):
def choices(population: Sequence[_T], weights: Optional[Sequence[float]] = ..., *, cum_weights: Optional[Sequence[float]] = ..., k: int = ...) -> List[_T]: ...
def shuffle(x: List[Any], random: Union[Callable[[], float], None] = ...) -> None: ...
if sys.version_info >= (3, 9):
def sample(population: Union[Sequence[_T], AbstractSet[_T]], k: int, *, counts: Optional[Iterable[_T]] = ...) -> List[_T]: ...
else:
def sample(population: Union[Sequence[_T], AbstractSet[_T]], k: int) -> List[_T]: ...
def random() -> float: ...
def uniform(a: float, b: float) -> float: ...
def triangular(low: float = ..., high: float = ...,
mode: Optional[float] = ...) -> float: ...
def betavariate(alpha: float, beta: float) -> float: ...
def expovariate(lambd: float) -> float: ...
def gammavariate(alpha: float, beta: float) -> float: ...
def gauss(mu: float, sigma: float) -> float: ...
def lognormvariate(mu: float, sigma: float) -> float: ...
def normalvariate(mu: float, sigma: float) -> float: ...
def vonmisesvariate(mu: float, kappa: float) -> float: ...
def paretovariate(alpha: float) -> float: ...
def weibullvariate(alpha: float, beta: float) -> float: ...
@@ -0,0 +1,89 @@
# Stubs for tracemalloc (Python 3.4+)
import sys
from typing import List, Optional, Sequence, Tuple, Union, overload
from _tracemalloc import *
def get_object_traceback(obj: object) -> Optional[Traceback]: ...
def take_snapshot() -> Snapshot: ...
if sys.version_info >= (3, 6):
class DomainFilter:
inclusive: bool
domain: int
def __init__(self, inclusive: bool, domain: int) -> None: ...
class Filter:
if sys.version_info >= (3, 6):
domain: Optional[int]
inclusive: bool
lineno: Optional[int]
filename_pattern: str
all_frames: bool
if sys.version_info >= (3, 6):
def __init__(self, inclusive: bool, filename_pattern: str, lineno: Optional[int] = ..., all_frames: bool = ..., domain: Optional[int] = ...) -> None: ...
else:
def __init__(self, inclusive: bool, filename_pattern: str, lineno: Optional[int] = ..., all_frames: bool = ...) -> None: ...
class Statistic:
count: int
size: int
traceback: Traceback
def __init__(self, traceback: Traceback, size: int, count: int) -> None: ...
class StatisticDiff:
count: int
count_diff: int
size: int
size_diff: int
traceback: Traceback
def __init__(self, traceback: Traceback, size: int, size_diff: int, count: int, count_diff: int) -> None: ...
_FrameTupleT = Tuple[str, int]
class Frame:
filename: str
lineno: int
def __init__(self, frame: _FrameTupleT) -> None: ...
if sys.version_info >= (3, 9):
_TraceTupleT = Union[Tuple[int, int, Sequence[_FrameTupleT], Optional[int]], Tuple[int, int, Sequence[_FrameTupleT]]]
else:
_TraceTupleT = Tuple[int, int, Sequence[_FrameTupleT]]
class Trace:
domain: int
size: int
traceback: Traceback
def __init__(self, trace: _TraceTupleT) -> None: ...
class Traceback(Sequence[Frame]):
if sys.version_info >= (3, 9):
total_nframe: Optional[int]
def __init__(self, frames: Sequence[_FrameTupleT], total_nframe: Optional[int] = ...) -> None: ...
else:
def __init__(self, frames: Sequence[_FrameTupleT]) -> None: ...
if sys.version_info >= (3, 7):
def format(self, limit: Optional[int] = ..., most_recent_first: bool = ...) -> List[str]: ...
else:
def format(self, limit: Optional[int] = ...) -> List[str]: ...
@overload
def __getitem__(self, i: int) -> Frame: ...
@overload
def __getitem__(self, s: slice) -> Sequence[Frame]: ...
def __len__(self) -> int: ...
class Snapshot:
def __init__(self, traces: Sequence[_TraceTupleT], traceback_limit: int) -> None: ...
def compare_to(self, old_snapshot: Snapshot, key_type: str, cumulative: bool = ...) -> List[StatisticDiff]: ...
def dump(self, filename: str) -> None: ...
if sys.version_info >= (3, 6):
def filter_traces(self, filters: Sequence[Union[DomainFilter, Filter]]) -> Snapshot: ...
else:
def filter_traces(self, filters: Sequence[Filter]) -> Snapshot: ...
@staticmethod
def load(filename: str) -> Snapshot: ...
def statistics(self, key_type: str, cumulative: bool = ...) -> List[Statistic]: ...
traceback_limit: int
traces: Sequence[Trace]
@@ -0,0 +1,99 @@
import sys
from typing import Optional, Type, Union, Tuple, Any
from types import TracebackType
_KeyType = Union[HKEYType, int]
def CloseKey(__hkey: _KeyType): ...
def ConnectRegistry(__computer_name: Optional[str], __key: _KeyType) -> HKEYType: ...
def CreateKey(__key: _KeyType, __sub_key: Optional[str]) -> HKEYType: ...
def CreateKeyEx(key: _KeyType, sub_key: Optional[str], reserved: int = ..., access: int = ...) -> HKEYType: ...
def DeleteKey(__key: _KeyType, __sub_key: str) -> None: ...
def DeleteKeyEx(key: _KeyType, sub_key: str, access: int = ..., reserved: int = ...) -> None: ...
def DeleteValue(__key: _KeyType, __value: str) -> None: ...
def EnumKey(__key: _KeyType, __index: int) -> str: ...
def EnumValue(__key: _KeyType, __index: int) -> Tuple[str, Any, int]: ...
def ExpandEnvironmentStrings(__str: str) -> str: ...
def FlushKey(__key: _KeyType) -> None: ...
def LoadKey(__key: _KeyType, __sub_key: str, __file_name: str) -> None: ...
def OpenKey(key: _KeyType, sub_key: str, reserved: int = ..., access: int = ...) -> HKEYType: ...
def OpenKeyEx(key: _KeyType, sub_key: str, reserved: int = ..., access: int = ...) -> HKEYType: ...
def QueryInfoKey(__key: _KeyType) -> Tuple[int, int, int]: ...
def QueryValue(__key: _KeyType, __sub_key: Optional[str]) -> str: ...
def QueryValueEx(__key: _KeyType, __name: str) -> Tuple[Any, int]: ...
def SaveKey(__key: _KeyType, __file_name: str) -> None: ...
def SetValue(__key: _KeyType, __sub_key: str, __type: int, __value: str) -> None: ...
def SetValueEx(__key: _KeyType, __value_name: str, __reserved: Any, __type: int, __value: str) -> None: ... # reserved is ignored
def DisableReflectionKey(__key: _KeyType) -> None: ...
def EnableReflectionKey(__key: _KeyType) -> None: ...
def QueryReflectionKey(__key: _KeyType) -> bool: ...
HKEY_CLASSES_ROOT: int
HKEY_CURRENT_USER: int
HKEY_LOCAL_MACHINE: int
HKEY_USERS: int
HKEY_PERFORMANCE_DATA: int
HKEY_CURRENT_CONFIG: int
HKEY_DYN_DATA: int
KEY_ALL_ACCESS: int
KEY_WRITE: int
KEY_READ: int
KEY_EXECUTE: int
KEY_QUERY_VALUE: int
KEY_SET_VALUE: int
KEY_CREATE_SUB_KEY: int
KEY_ENUMERATE_SUB_KEYS: int
KEY_NOTIFY: int
KEY_CREATE_LINK: int
KEY_WOW64_64KEY: int
KEY_WOW64_32KEY: int
REG_BINARY: int
REG_DWORD: int
REG_DWORD_LITTLE_ENDIAN: int
REG_DWORD_BIG_ENDIAN: int
REG_EXPAND_SZ: int
REG_LINK: int
REG_MULTI_SZ: int
REG_NONE: int
if sys.version_info >= (3, 6):
REG_QWORD: int
REG_QWORD_LITTLE_ENDIAN: int
REG_RESOURCE_LIST: int
REG_FULL_RESOURCE_DESCRIPTOR: int
REG_RESOURCE_REQUIREMENTS_LIST: int
REG_SZ: int
REG_CREATED_NEW_KEY: int # undocumented
REG_LEGAL_CHANGE_FILTER: int # undocumented
REG_LEGAL_OPTION: int # undocumented
REG_NOTIFY_CHANGE_ATTRIBUTES: int # undocumented
REG_NOTIFY_CHANGE_LAST_SET: int # undocumented
REG_NOTIFY_CHANGE_NAME: int # undocumented
REG_NOTIFY_CHANGE_SECURITY: int # undocumented
REG_NO_LAZY_FLUSH: int # undocumented
REG_OPENED_EXISTING_KEY: int # undocumented
REG_OPTION_BACKUP_RESTORE: int # undocumented
REG_OPTION_CREATE_LINK: int # undocumented
REG_OPTION_NON_VOLATILE: int # undocumented
REG_OPTION_OPEN_LINK: int # undocumented
REG_OPTION_RESERVED: int # undocumented
REG_OPTION_VOLATILE: int # undocumented
REG_REFRESH_HIVE: int # undocumented
REG_WHOLE_HIVE_VOLATILE: int # undocumented
error = OSError
# Though this class has a __name__ of PyHKEY, it's exposed as HKEYType for some reason
class HKEYType:
def __bool__(self) -> bool: ...
def __int__(self) -> int: ...
def __enter__(self) -> HKEYType: ...
def __exit__(self, exc_type: Optional[Type[BaseException]], exc_val: Optional[BaseException], exc_tb: Optional[TracebackType]) -> Optional[bool]: ...
def Close(self) -> None: ...
def Detach(self) -> int: ...
@@ -0,0 +1,283 @@
import io
import sys
import time
import gzip
import http.client
from typing import Any, Callable, Dict, IO, Iterable, List, Mapping, Optional, Protocol, Text, Tuple, Type, TypeVar, Union, overload
from types import TracebackType
from datetime import datetime
if sys.version_info >= (3, 8):
from typing import Literal
else:
from typing_extensions import Literal
_T = TypeVar("_T")
class _HasTimeTuple(Protocol):
def timetuple(self) -> time.struct_time: ...
class _HasWrite(Protocol):
def write(self, __o: str) -> None: ...
class _HasRead(Protocol):
def read(self) -> bytes: ...
_DateTimeComparable = Union[DateTime, datetime, str, _HasTimeTuple]
_Marshallable = Union[None, bool, int, float, str, bytes, tuple, list, dict, datetime, DateTime, Binary]
_XMLDate = Union[int, datetime, Tuple[int, ...], time.struct_time]
_HostType = Union[Tuple[str, Dict[str, str]], str]
def escape(s: str) -> str: ... # undocumented
PARSE_ERROR: int # undocumented
SERVER_ERROR: int # undocumented
APPLICATION_ERROR: int # undocumented
SYSTEM_ERROR: int # undocumented
TRANSPORT_ERROR: int # undocumented
NOT_WELLFORMED_ERROR: int # undocumented
UNSUPPORTED_ENCODING: int # undocumented
INVALID_ENCODING_CHAR: int # undocumented
INVALID_XMLRPC: int # undocumented
METHOD_NOT_FOUND: int # undocumented
INVALID_METHOD_PARAMS: int # undocumented
INTERNAL_ERROR: int # undocumented
class Error(Exception): ...
class ProtocolError(Error):
url: str
errcode: int
errmsg: str
headers: Dict[str, str]
def __init__(self, url: str, errcode: int, errmsg: str, headers: Dict[str, str]) -> None: ...
class ResponseError(Error): ...
class Fault(Error):
faultCode: str
faultString: str
def __init__(self, faultCode: str, faultString: str, **extra: Any) -> None: ...
boolean = bool
Boolean = bool
def _iso8601_format(value: datetime) -> str: ... # undocumented
def _strftime(value: _XMLDate) -> str: ... # undocumented
class DateTime:
value: str # undocumented
def __init__(self, value: Union[int, str, datetime, time.struct_time, Tuple[int, ...]] = ...): ...
def __lt__(self, other: _DateTimeComparable) -> bool: ...
def __le__(self, other: _DateTimeComparable) -> bool: ...
def __gt__(self, other: _DateTimeComparable) -> bool: ...
def __ge__(self, other: _DateTimeComparable) -> bool: ...
def __eq__(self, other: _DateTimeComparable) -> bool: ... # type: ignore
def make_comparable(self, other: _DateTimeComparable) -> Tuple[str, str]: ... # undocumented
def timetuple(self) -> time.struct_time: ... # undocumented
def decode(self, data: Any) -> None: ...
def encode(self, out: _HasWrite) -> None: ...
def _datetime(data: Any) -> DateTime: ... # undocumented
def _datetime_type(data: str) -> datetime: ... # undocumented
class Binary:
data: bytes
def __init__(self, data: Optional[bytes] = ...) -> None: ...
def decode(self, data: bytes) -> None: ...
def encode(self, out: _HasWrite) -> None: ...
def _binary(data: bytes) -> Binary: ... # undocumented
WRAPPERS: Tuple[Type[DateTime], Type[Binary]] # undocumented
class ExpatParser: # undocumented
def __init__(self, target: Unmarshaller) -> None: ...
def feed(self, data: Union[Text, bytes]) -> None: ...
def close(self) -> None: ...
class Marshaller:
dispatch: Dict[Type[Any], Callable[[Marshaller, Any, Callable[[str], Any]], None]] = ... # TODO: Replace 'Any' with some kind of binding
memo: Dict[Any, None]
data: None
encoding: Optional[str]
allow_none: bool
def __init__(self, encoding: Optional[str] = ..., allow_none: bool = ...) -> None: ...
def dumps(self, values: Union[Fault, Iterable[_Marshallable]]) -> str: ...
def __dump(self, value: Union[_Marshallable], write: Callable[[str], Any]) -> None: ... # undocumented
def dump_nil(self, value: None, write: Callable[[str], Any]) -> None: ...
def dump_bool(self, value: bool, write: Callable[[str], Any]) -> None: ...
def dump_long(self, value: int, write: Callable[[str], Any]) -> None: ...
def dump_int(self, value: int, write: Callable[[str], Any]) -> None: ...
def dump_double(self, value: float, write: Callable[[str], Any]) -> None: ...
def dump_unicode(self, value: str, write: Callable[[str], Any], escape: Callable[[str], str] = ...) -> None: ...
def dump_bytes(self, value: bytes, write: Callable[[str], Any]) -> None: ...
def dump_array(self, value: Iterable[_Marshallable], write: Callable[[str], Any]) -> None: ...
def dump_struct(self, value: Mapping[str, _Marshallable], write: Callable[[str], Any], escape: Callable[[str], str] = ...) -> None: ...
def dump_datetime(self, value: _XMLDate, write: Callable[[str], Any]) -> None: ...
def dump_instance(self, value: object, write: Callable[[str], Any]) -> None: ...
class Unmarshaller:
dispatch: Dict[str, Callable[[Unmarshaller, str], None]] = ...
_type: Optional[str]
_stack: List[_Marshallable]
_marks: List[int]
_data: List[str]
_value: bool
_methodname: Optional[str]
_encoding: str
append: Callable[[Any], None]
_use_datetime: bool
_use_builtin_types: bool
def __init__(self, use_datetime: bool = ..., use_builtin_types: bool = ...) -> None: ...
def close(self) -> Tuple[_Marshallable, ...]: ...
def getmethodname(self) -> Optional[str]: ...
def xml(self, encoding: str, standalone: Any) -> None: ... # Standalone is ignored
def start(self, tag: str, attrs: Dict[str, str]) -> None: ...
def data(self, text: str) -> None: ...
def end(self, tag: str) -> None: ...
def end_dispatch(self, tag: str, data: str) -> None: ...
def end_nil(self, data: str) -> None: ...
def end_boolean(self, data: str) -> None: ...
def end_int(self, data: str) -> None: ...
def end_double(self, data: str) -> None: ...
if sys.version_info >= (3, 6):
def end_bigdecimal(self, data: str) -> None: ...
def end_string(self, data: str) -> None: ...
def end_array(self, data: str) -> None: ...
def end_struct(self, data: str) -> None: ...
def end_base64(self, data: str) -> None: ...
def end_dateTime(self, data: str) -> None: ...
def end_value(self, data: str) -> None: ...
def end_params(self, data: str) -> None: ...
def end_fault(self, data: str) -> None: ...
def end_methodName(self, data: str) -> None: ...
class _MultiCallMethod: # undocumented
__call_list: List[Tuple[str, Tuple[_Marshallable, ...]]]
__name: str
def __init__(self, call_list: List[Tuple[str, _Marshallable]], name: str) -> None: ...
def __getattr__(self, name: str) -> _MultiCallMethod: ...
def __call__(self, *args: _Marshallable) -> None: ...
class MultiCallIterator: # undocumented
results: List[List[_Marshallable]]
def __init__(self, results: List[List[_Marshallable]]) -> None: ...
def __getitem__(self, i: int) -> _Marshallable: ...
class MultiCall:
__server: ServerProxy
__call_list: List[Tuple[str, Tuple[_Marshallable, ...]]]
def __init__(self, server: ServerProxy) -> None: ...
def __getattr__(self, item: str) -> _MultiCallMethod: ...
def __call__(self) -> MultiCallIterator: ...
# A little white lie
FastMarshaller: Optional[Marshaller]
FastParser: Optional[ExpatParser]
FastUnmarshaller: Optional[Unmarshaller]
def getparser(use_datetime: bool = ..., use_builtin_types: bool = ...) -> Tuple[ExpatParser, Unmarshaller]: ...
def dumps(params: Union[Fault, Tuple[_Marshallable, ...]], methodname: Optional[str] = ..., methodresponse: Optional[bool] = ..., encoding: Optional[str] = ..., allow_none: bool = ...) -> str: ...
def loads(data: str, use_datetime: bool = ..., use_builtin_types: bool = ...) -> Tuple[Tuple[_Marshallable, ...], Optional[str]]: ...
def gzip_encode(data: bytes) -> bytes: ... # undocumented
def gzip_decode(data: bytes, max_decode: int = ...) -> bytes: ... # undocumented
class GzipDecodedResponse(gzip.GzipFile): # undocumented
io: io.BytesIO
def __init__(self, response: _HasRead) -> None: ...
def close(self) -> None: ...
class _Method: # undocumented
__send: Callable[[str, Tuple[_Marshallable, ...]], _Marshallable]
__name: str
def __init__(self, send: Callable[[str, Tuple[_Marshallable, ...]], _Marshallable], name: str) -> None: ...
def __getattr__(self, name: str) -> _Method: ...
def __call__(self, *args: _Marshallable) -> _Marshallable: ...
class Transport:
user_agent: str = ...
accept_gzip_encoding: bool = ...
encode_threshold: Optional[int] = ...
_use_datetime: bool
_use_builtin_types: bool
_connection: Tuple[Optional[_HostType], Optional[http.client.HTTPConnection]]
_headers: List[Tuple[str, str]]
_extra_headers: List[Tuple[str, str]]
if sys.version_info >= (3, 8):
def __init__(self, use_datetime: bool = ..., use_builtin_types: bool = ..., *, headers: Iterable[Tuple[str, str]] = ...) -> None: ...
else:
def __init__(self, use_datetime: bool = ..., use_builtin_types: bool = ...) -> None: ...
def request(self, host: _HostType, handler: str, request_body: bytes, verbose: bool = ...) -> Tuple[_Marshallable, ...]: ...
def single_request(self, host: _HostType, handler: str, request_body: bytes, verbose: bool = ...) -> Tuple[_Marshallable, ...]: ...
def getparser(self) -> Tuple[ExpatParser, Unmarshaller]: ...
def get_host_info(self, host: _HostType) -> Tuple[str, List[Tuple[str, str]], Dict[str, str]]: ...
def make_connection(self, host: _HostType) -> http.client.HTTPConnection: ...
def close(self) -> None: ...
def send_request(self, host: _HostType, handler: str, request_body: bytes, debug: bool) -> http.client.HTTPConnection: ...
def send_headers(self, connection: http.client.HTTPConnection, headers: List[Tuple[str, str]]) -> None: ...
def send_content(self, connection: http.client.HTTPConnection, request_body: bytes) -> None: ...
def parse_response(self, response: http.client.HTTPResponse) -> Tuple[_Marshallable, ...]: ...
class SafeTransport(Transport):
if sys.version_info >= (3, 8):
def __init__(self, use_datetime: bool = ..., use_builtin_types: bool = ..., *, headers: Iterable[Tuple[str, str]] = ..., context: Optional[Any] = ...) -> None: ...
else:
def __init__(self, use_datetime: bool = ..., use_builtin_types: bool = ..., *, context: Optional[Any] = ...) -> None: ...
def make_connection(self, host: _HostType) -> http.client.HTTPSConnection: ...
class ServerProxy:
__host: str
__handler: str
__transport: Transport
__encoding: str
__verbose: bool
__allow_none: bool
if sys.version_info >= (3, 8):
def __init__(self, uri, transport: Optional[Transport] = ..., encoding: Optional[str] = ..., verbose: bool = ..., allow_none: bool = ..., use_datetime: bool = ..., use_builtin_types: bool = ..., *, headers: Iterable[Tuple[str, str]] = ..., context: Optional[Any] = ...) -> None: ...
else:
def __init__(self, uri, transport: Optional[Transport] = ..., encoding: Optional[str] = ..., verbose: bool = ..., allow_none: bool = ..., use_datetime: bool = ..., use_builtin_types: bool = ..., *, context: Optional[Any] = ...) -> None: ...
def __getattr__(self, name: str) -> _Method: ...
@overload
def __call__(self, attr: Literal["close"]) -> Callable[[], None]: ...
@overload
def __call__(self, attr: Literal["transport"]) -> Transport: ...
@overload
def __call__(self, attr: str) -> Union[Callable[[], None], Transport]: ...
def __enter__(self) -> ServerProxy: ...
def __exit__(self, exc_type: Optional[Type[BaseException]], exc_val: Optional[BaseException], exc_tb: Optional[TracebackType]) -> None: ...
def __close(self) -> None: ... # undocumented
def __request(self, methodname: str, params: Tuple[_Marshallable, ...]) -> Tuple[_Marshallable, ...]: ... # undocumented
Server = ServerProxy
@@ -0,0 +1,105 @@
import http.server
import socketserver
import pydoc
import sys
from xmlrpc.client import Fault
from typing import Any, Callable, Dict, Iterable, List, Mapping, Optional, Pattern, Protocol, Tuple, Type, Union
from datetime import datetime
_Marshallable = Union[None, bool, int, float, str, bytes, tuple, list, dict, datetime]
class _DispatchProtocol(Protocol):
def __call__(self, *args: _Marshallable) -> _Marshallable: ...
def resolve_dotted_attribute(obj: Any, attr: str, allow_dotted_names: bool = ...) -> Any: ... # undocumented
def list_public_methods(obj: Any) -> List[str]: ... # undocumented
class SimpleXMLRPCDispatcher: # undocumented
funcs: Dict[str, _DispatchProtocol]
instance: Optional[Any]
allow_none: bool
encoding: str
use_builtin_types: bool
def __init__(self, allow_none: bool = ..., encoding: Optional[str] = ..., use_builtin_types: bool = ...) -> None: ...
def register_instance(self, instance: Any, allow_dotted_names: bool = ...) -> None: ...
if sys.version_info >= (3, 7):
def register_function(self, function: Optional[_DispatchProtocol] = ..., name: Optional[str] = ...) -> Callable[..., Any]: ...
else:
def register_function(self, function: _DispatchProtocol, name: Optional[str] = ...) -> Callable[..., Any]: ...
def register_introspection_functions(self) -> None: ...
def register_multicall_functions(self) -> None: ...
def _marshaled_dispatch(self, data: str, dispatch_method: Optional[Callable[[Optional[str], Tuple[_Marshallable, ...]], Union[Fault, Tuple[_Marshallable, ...]]]] = ..., path: Optional[Any] = ...) -> str: ... # undocumented
def system_listMethods(self) -> List[str]: ... # undocumented
def system_methodSignature(self, method_name: str) -> str: ... # undocumented
def system_methodHelp(self, method_name: str) -> str: ... # undocumented
def system_multicall(self, call_list: List[Dict[str, _Marshallable]]) -> List[_Marshallable]: ... # undocumented
def _dispatch(self, method: str, params: Iterable[_Marshallable]) -> _Marshallable: ... # undocumented
class SimpleXMLRPCRequestHandler(http.server.BaseHTTPRequestHandler):
rpc_paths: Tuple[str, str] = ...
encode_threshold: int = ... # undocumented
aepattern: Pattern[str] # undocumented
def accept_encodings(self) -> Dict[str, float]: ...
def is_rpc_path_valid(self) -> bool: ...
def do_POST(self) -> None: ...
def decode_request_content(self, data: bytes) -> Optional[bytes]: ...
def report_404(self) -> None: ...
def log_request(self, code: Union[int, str] = ..., size: Union[int, str] = ...) -> None: ...
class SimpleXMLRPCServer(socketserver.TCPServer, SimpleXMLRPCDispatcher):
allow_reuse_address: bool = ...
_send_traceback_handler: bool = ...
def __init__(self, addr: Tuple[str, int], requestHandler: Type[SimpleXMLRPCRequestHandler] = ..., logRequests: bool = ..., allow_none: bool = ..., encoding: Optional[str] = ..., bind_and_activate: bool = ..., use_builtin_types: bool = ...) -> None: ...
class MultiPathXMLRPCServer(SimpleXMLRPCServer): # undocumented
dispatchers: Dict[str, SimpleXMLRPCDispatcher]
allow_none: bool
encoding: str
def __init__(self, addr: Tuple[str, int], requestHandler: Type[SimpleXMLRPCRequestHandler] = ..., logRequests: bool = ..., allow_none: bool = ..., encoding: Optional[str] = ..., bind_and_activate: bool = ..., use_builtin_types: bool = ...) -> None: ...
def add_dispatcher(self, path: str, dispatcher: SimpleXMLRPCDispatcher) -> SimpleXMLRPCDispatcher: ...
def get_dispatcher(self, path: str) -> SimpleXMLRPCDispatcher: ...
def _marshaled_dispatch(self, data: str, dispatch_method: Optional[Callable[[Optional[str], Tuple[_Marshallable, ...]], Union[Fault, Tuple[_Marshallable, ...]]]] = ..., path: Optional[Any] = ...) -> str: ...
class CGIXMLRPCRequestHandler(SimpleXMLRPCDispatcher):
def __init__(self, allow_none: bool = ..., encoding: Optional[str] = ..., use_builtin_types: bool = ...) -> None: ...
def handle_xmlrpc(self, request_text: str) -> None: ...
def handle_get(self) -> None: ...
def handle_request(self, request_text: Optional[str] = ...) -> None: ...
class ServerHTMLDoc(pydoc.HTMLDoc): # undocumented
def docroutine(self, object: object, name: str, mod: Optional[str] = ..., funcs: Mapping[str, str] = ..., classes: Mapping[str, str] = ..., methods: Mapping[str, str] = ..., cl: Optional[type] = ...) -> str: ... # type: ignore
def docserver(self, server_name: str, package_documentation: str, methods: Dict[str, str]) -> str: ...
class XMLRPCDocGenerator: # undocumented
server_name: str
server_documentation: str
server_title: str
def __init__(self) -> None: ...
def set_server_title(self, server_title: str) -> None: ...
def set_server_name(self, server_name: str) -> None: ...
def set_server_documentation(self, server_documentation: str) -> None: ...
def generate_html_documentation(self) -> str: ...
class DocXMLRPCRequestHandler(SimpleXMLRPCRequestHandler):
def do_GET(self) -> None: ...
class DocXMLRPCServer(SimpleXMLRPCServer, XMLRPCDocGenerator):
def __init__(self, addr: Tuple[str, int], requestHandler: Type[SimpleXMLRPCRequestHandler] = ..., logRequests: bool = ..., allow_none: bool = ..., encoding: Optional[str] = ..., bind_and_activate: bool = ..., use_builtin_types: bool = ...) -> None: ...
class DocCGIXMLRPCRequestHandler(CGIXMLRPCRequestHandler, XMLRPCDocGenerator):
def __init__(self) -> None: ...
@@ -0,0 +1,22 @@
import sys
from typing import Any, AnyStr, Callable, ContextManager, Generic, IO, Optional, Text, Type, Union
if sys.version_info >= (3, 6):
from os import PathLike
_Path = Union[str, bytes, PathLike[str], PathLike[bytes]]
else:
_Path = Union[Text, bytes]
def replace_atomic(src: AnyStr, dst: AnyStr) -> None: ...
def move_atomic(src: AnyStr, dst: AnyStr) -> None: ...
class AtomicWriter(object):
def __init__(self, path: _Path, mode: Text = ..., overwrite: bool = ...) -> None: ...
def open(self) -> ContextManager[IO[Any]]: ...
def _open(self, get_fileobject: Callable[..., IO[AnyStr]]) -> ContextManager[IO[AnyStr]]: ...
def get_fileobject(self, dir: Optional[_Path] = ..., **kwargs: Any) -> IO[Any]: ...
def sync(self, f: IO[Any]) -> None: ...
def commit(self, f: IO[Any]) -> None: ...
def rollback(self, f: IO[Any]) -> None: ...
def atomic_write(
path: _Path, writer_cls: Type[AtomicWriter] = ..., **cls_kwargs: object,
) -> ContextManager[IO[Any]]: ...
@@ -0,0 +1,26 @@
import sys
from typing import Any, Tuple
from .universaldetector import UniversalDetector as UniversalDetector
def __getattr__(name: str) -> Any: ... # incomplete
if sys.version_info >= (3, 8):
from typing import TypedDict
else:
from typing_extensions import TypedDict
class _LangModelType(TypedDict):
char_to_order_map: Tuple[int, ...]
precedence_matrix: Tuple[int, ...]
typical_positive_ratio: float
keep_english_letter: bool
charset_name: str
language: str
class _SMModelType(TypedDict):
class_table: Tuple[int, ...]
class_factor: int
state_table: Tuple[int, ...]
char_len_table: Tuple[int, ...]
name: str
@@ -0,0 +1,40 @@
class InputState(object):
PURE_ASCII: int
ESC_ASCII: int
HIGH_BYTE: int
class LanguageFilter(object):
CHINESE_SIMPLIFIED: int
CHINESE_TRADITIONAL: int
JAPANESE: int
KOREAN: int
NON_CJK: int
ALL: int
CHINESE: int
CJK: int
class ProbingState(object):
DETECTING: int
FOUND_IT: int
NOT_ME: int
class MachineState(object):
START: int
ERROR: int
ITS_ME: int
class SequenceLikelihood(object):
NEGATIVE: int
UNLIKELY: int
LIKELY: int
POSITIVE: int
@classmethod
def get_num_categories(cls) -> int: ...
class CharacterCategory(object):
UNDEFINED: int
LINE_BREAK: int
SYMBOL: int
DIGIT: int
CONTROL: int
@@ -0,0 +1,8 @@
from typing import Tuple
from . import _LangModelType
Latin5_BulgarianCharToOrderMap: Tuple[int, ...]
win1251BulgarianCharToOrderMap: Tuple[int, ...]
BulgarianLangModel: Tuple[int, ...]
Latin5BulgarianModel: _LangModelType
Win1251BulgarianModel: _LangModelType
@@ -0,0 +1,16 @@
from typing import Tuple
from . import _LangModelType
KOI8R_char_to_order_map: Tuple[int, ...]
win1251_char_to_order_map: Tuple[int, ...]
latin5_char_to_order_map: Tuple[int, ...]
macCyrillic_char_to_order_map: Tuple[int, ...]
IBM855_char_to_order_map: Tuple[int, ...]
IBM866_char_to_order_map: Tuple[int, ...]
RussianLangModel: Tuple[int, ...]
Koi8rModel: _LangModelType
Win1251CyrillicModel: _LangModelType
Latin5CyrillicModel: _LangModelType
MacCyrillicModel: _LangModelType
Ibm866Model: _LangModelType
Ibm855Model: _LangModelType
@@ -0,0 +1,8 @@
from typing import Tuple
from . import _LangModelType
Latin7_char_to_order_map: Tuple[int, ...]
win1253_char_to_order_map: Tuple[int, ...]
GreekLangModel: Tuple[int, ...]
Latin7GreekModel: _LangModelType
Win1253GreekModel: _LangModelType
@@ -0,0 +1,6 @@
from typing import Tuple
from . import _LangModelType
WIN1255_CHAR_TO_ORDER_MAP: Tuple[int, ...]
HEBREW_LANG_MODEL: Tuple[int, ...]
Win1255HebrewModel: _LangModelType
@@ -0,0 +1,8 @@
from typing import Tuple
from . import _LangModelType
Latin2_HungarianCharToOrderMap: Tuple[int, ...]
win1250HungarianCharToOrderMap: Tuple[int, ...]
HungarianLangModel: Tuple[int, ...]
Latin2HungarianModel: _LangModelType
Win1250HungarianModel: _LangModelType
@@ -0,0 +1,6 @@
from typing import Tuple
from . import _LangModelType
TIS620CharToOrderMap: Tuple[int, ...]
ThaiLangModel: Tuple[int, ...]
TIS620ThaiModel: _LangModelType
@@ -0,0 +1,6 @@
from typing import Tuple
from . import _LangModelType
Latin5_TurkishCharToOrderMap: Tuple[int, ...]
TurkishLangModel: Tuple[int, ...]
Latin5TurkishModel: _LangModelType
@@ -0,0 +1,31 @@
import sys
from typing import Dict, Pattern, Optional
from typing_extensions import TypedDict
from logging import Logger
class _FinalResultType(TypedDict):
encoding: str
confidence: float
language: str
class _IntermediateResultType(TypedDict):
encoding: Optional[str]
confidence: float
language: Optional[str]
class UniversalDetector(object):
MINIMUM_THRESHOLD: float
HIGH_BYTE_DETECTOR: Pattern[bytes]
ESC_DETECTOR: Pattern[bytes]
WIN_BYTE_DETECTOR: Pattern[bytes]
ISO_WIN_MAP: Dict[str, str]
result: _IntermediateResultType
done: bool
lang_filter: int
logger: Logger
def __init__(self, lang_filter: int = ...) -> None: ...
def reset(self) -> None: ...
def feed(self, byte_str: bytes) -> None: ...
def close(self) -> _FinalResultType: ...
@@ -0,0 +1,4 @@
from typing import List
__version__: str
VERSION: List[str]
@@ -0,0 +1,5 @@
from pytz import BaseTzInfo
def reload_localzone() -> None: ...
def get_localzone() -> BaseTzInfo: ...
@@ -0,0 +1,265 @@
# Stubs for pkg_resources (Python 3.4)
from typing import Any, Callable, Dict, IO, Iterable, Generator, Optional, Sequence, Tuple, List, Set, Union, TypeVar, overload
from abc import ABCMeta
import importlib.abc
import types
import zipimport
_T = TypeVar("_T")
_NestedStr = Union[str, Iterable[Union[str, Iterable[Any]]]]
_InstallerType = Callable[[Requirement], Optional[Distribution]]
_EPDistType = Union[Distribution, Requirement, str]
_MetadataType = Optional[IResourceProvider]
_PkgReqType = Union[str, Requirement]
_DistFinderType = Callable[[_Importer, str, bool], Generator[Distribution, None, None]]
_NSHandlerType = Callable[[_Importer, str, str, types.ModuleType], str]
def declare_namespace(name: str) -> None: ...
def fixup_namespace_packages(path_item: str) -> None: ...
class WorkingSet:
entries: List[str]
def __init__(self, entries: Optional[Iterable[str]] = ...) -> None: ...
def require(self, *requirements: _NestedStr) -> Sequence[Distribution]: ...
def run_script(self, requires: str, script_name: str) -> None: ...
def iter_entry_points(self, group: str, name: Optional[str] = ...) -> Generator[EntryPoint, None, None]: ...
def add_entry(self, entry: str) -> None: ...
def __contains__(self, dist: Distribution) -> bool: ...
def __iter__(self) -> Generator[Distribution, None, None]: ...
def find(self, req: Requirement) -> Optional[Distribution]: ...
def resolve(
self, requirements: Iterable[Requirement], env: Optional[Environment] = ..., installer: Optional[_InstallerType] = ...
) -> List[Distribution]: ...
def add(self, dist: Distribution, entry: Optional[str] = ..., insert: bool = ..., replace: bool = ...) -> None: ...
def subscribe(self, callback: Callable[[Distribution], None]) -> None: ...
def find_plugins(
self, plugin_env: Environment, full_env: Optional[Environment] = ..., fallback: bool = ...
) -> Tuple[List[Distribution], Dict[Distribution, Exception]]: ...
working_set: WorkingSet
def require(*requirements: Union[str, Sequence[str]]) -> Sequence[Distribution]: ...
def run_script(requires: str, script_name: str) -> None: ...
def iter_entry_points(group: str, name: Optional[str] = ...) -> Generator[EntryPoint, None, None]: ...
def add_activation_listener(callback: Callable[[Distribution], None]) -> None: ...
class Environment:
def __init__(
self, search_path: Optional[Sequence[str]] = ..., platform: Optional[str] = ..., python: Optional[str] = ...
) -> None: ...
def __getitem__(self, project_name: str) -> List[Distribution]: ...
def __iter__(self) -> Generator[str, None, None]: ...
def add(self, dist: Distribution) -> None: ...
def remove(self, dist: Distribution) -> None: ...
def can_add(self, dist: Distribution) -> bool: ...
def __add__(self, other: Union[Distribution, Environment]) -> Environment: ...
def __iadd__(self, other: Union[Distribution, Environment]) -> Environment: ...
@overload
def best_match(self, req: Requirement, working_set: WorkingSet) -> Distribution: ...
@overload
def best_match(self, req: Requirement, working_set: WorkingSet, installer: Callable[[Requirement], _T] = ...) -> _T: ...
@overload
def obtain(self, requirement: Requirement) -> None: ...
@overload
def obtain(self, requirement: Requirement, installer: Callable[[Requirement], _T] = ...) -> _T: ...
def scan(self, search_path: Optional[Sequence[str]] = ...) -> None: ...
def parse_requirements(strs: Union[str, Iterable[str]]) -> Generator[Requirement, None, None]: ...
class Requirement:
unsafe_name: str
project_name: str
key: str
extras: Tuple[str, ...]
specs: List[Tuple[str, str]]
# TODO: change this to Optional[packaging.markers.Marker] once we can import
# packaging.markers
marker: Optional[Any]
@staticmethod
def parse(s: Union[str, Iterable[str]]) -> Requirement: ...
def __contains__(self, item: Union[Distribution, str, Tuple[str, ...]]) -> bool: ...
def __eq__(self, other_requirement: Any) -> bool: ...
def load_entry_point(dist: _EPDistType, group: str, name: str) -> None: ...
def get_entry_info(dist: _EPDistType, group: str, name: str) -> Optional[EntryPoint]: ...
@overload
def get_entry_map(dist: _EPDistType) -> Dict[str, Dict[str, EntryPoint]]: ...
@overload
def get_entry_map(dist: _EPDistType, group: str) -> Dict[str, EntryPoint]: ...
class EntryPoint:
name: str
module_name: str
attrs: Tuple[str, ...]
extras: Tuple[str, ...]
dist: Optional[Distribution]
def __init__(
self,
name: str,
module_name: str,
attrs: Tuple[str, ...] = ...,
extras: Tuple[str, ...] = ...,
dist: Optional[Distribution] = ...,
) -> None: ...
@classmethod
def parse(cls, src: str, dist: Optional[Distribution] = ...) -> EntryPoint: ...
@classmethod
def parse_group(
cls, group: str, lines: Union[str, Sequence[str]], dist: Optional[Distribution] = ...
) -> Dict[str, EntryPoint]: ...
@classmethod
def parse_map(
cls, data: Union[Dict[str, Union[str, Sequence[str]]], str, Sequence[str]], dist: Optional[Distribution] = ...
) -> Dict[str, EntryPoint]: ...
def load(self, require: bool = ..., env: Optional[Environment] = ..., installer: Optional[_InstallerType] = ...) -> Any: ...
def require(self, env: Optional[Environment] = ..., installer: Optional[_InstallerType] = ...) -> None: ...
def resolve(self) -> Any: ...
def find_distributions(path_item: str, only: bool = ...) -> Generator[Distribution, None, None]: ...
def get_distribution(dist: Union[Requirement, str, Distribution]) -> Distribution: ...
class Distribution(IResourceProvider, IMetadataProvider):
PKG_INFO: str
location: str
project_name: str
key: str
extras: List[str]
version: str
parsed_version: Tuple[str, ...]
py_version: str
platform: Optional[str]
precedence: int
def __init__(
self,
location: Optional[str] = ...,
metadata: _MetadataType = ...,
project_name: Optional[str] = ...,
version: Optional[str] = ...,
py_version: str = ...,
platform: Optional[str] = ...,
precedence: int = ...,
) -> None: ...
@classmethod
def from_location(
cls, location: str, basename: str, metadata: _MetadataType = ..., **kw: Union[str, None, int]
) -> Distribution: ...
@classmethod
def from_filename(cls, filename: str, metadata: _MetadataType = ..., **kw: Union[str, None, int]) -> Distribution: ...
def activate(self, path: Optional[List[str]] = ...) -> None: ...
def as_requirement(self) -> Requirement: ...
def requires(self, extras: Tuple[str, ...] = ...) -> List[Requirement]: ...
def clone(self, **kw: Union[str, int, None]) -> Requirement: ...
def egg_name(self) -> str: ...
def __cmp__(self, other: Any) -> bool: ...
def get_entry_info(self, group: str, name: str) -> Optional[EntryPoint]: ...
@overload
def get_entry_map(self) -> Dict[str, Dict[str, EntryPoint]]: ...
@overload
def get_entry_map(self, group: str) -> Dict[str, EntryPoint]: ...
def load_entry_point(self, group: str, name: str) -> None: ...
EGG_DIST: int
BINARY_DIST: int
SOURCE_DIST: int
CHECKOUT_DIST: int
DEVELOP_DIST: int
def resource_exists(package_or_requirement: _PkgReqType, resource_name: str) -> bool: ...
def resource_stream(package_or_requirement: _PkgReqType, resource_name: str) -> IO[bytes]: ...
def resource_string(package_or_requirement: _PkgReqType, resource_name: str) -> bytes: ...
def resource_isdir(package_or_requirement: _PkgReqType, resource_name: str) -> bool: ...
def resource_listdir(package_or_requirement: _PkgReqType, resource_name: str) -> List[str]: ...
def resource_filename(package_or_requirement: _PkgReqType, resource_name: str) -> str: ...
def set_extraction_path(path: str) -> None: ...
def cleanup_resources(force: bool = ...) -> List[str]: ...
class IResourceManager:
def resource_exists(self, package_or_requirement: _PkgReqType, resource_name: str) -> bool: ...
def resource_stream(self, package_or_requirement: _PkgReqType, resource_name: str) -> IO[bytes]: ...
def resource_string(self, package_or_requirement: _PkgReqType, resource_name: str) -> bytes: ...
def resource_isdir(self, package_or_requirement: _PkgReqType, resource_name: str) -> bool: ...
def resource_listdir(self, package_or_requirement: _PkgReqType, resource_name: str) -> List[str]: ...
def resource_filename(self, package_or_requirement: _PkgReqType, resource_name: str) -> str: ...
def set_extraction_path(self, path: str) -> None: ...
def cleanup_resources(self, force: bool = ...) -> List[str]: ...
def get_cache_path(self, archive_name: str, names: Iterable[str] = ...) -> str: ...
def extraction_error(self) -> None: ...
def postprocess(self, tempname: str, filename: str) -> None: ...
@overload
def get_provider(package_or_requirement: str) -> IResourceProvider: ...
@overload
def get_provider(package_or_requirement: Requirement) -> Distribution: ...
class IMetadataProvider:
def has_metadata(self, name: str) -> bool: ...
def metadata_isdir(self, name: str) -> bool: ...
def metadata_listdir(self, name: str) -> List[str]: ...
def get_metadata(self, name: str) -> str: ...
def get_metadata_lines(self, name: str) -> Generator[str, None, None]: ...
def run_script(self, script_name: str, namespace: Dict[str, Any]) -> None: ...
class ResolutionError(Exception): ...
class DistributionNotFound(ResolutionError): ...
class VersionConflict(ResolutionError):
@property
def dist(self) -> Any: ...
@property
def req(self) -> Any: ...
def report(self) -> str: ...
def with_context(self, required_by: Set[Union[Distribution, str]]) -> VersionConflict: ...
class ContextualVersionConflict(VersionConflict):
@property
def required_by(self) -> Set[Union[Distribution, str]]: ...
class UnknownExtra(ResolutionError): ...
class ExtractionError(Exception):
manager: IResourceManager
cache_path: str
original_error: Exception
class _Importer(importlib.abc.MetaPathFinder, importlib.abc.InspectLoader, metaclass=ABCMeta): ...
def register_finder(importer_type: type, distribution_finder: _DistFinderType) -> None: ...
def register_loader_type(loader_type: type, provider_factory: Callable[[types.ModuleType], IResourceProvider]) -> None: ...
def register_namespace_handler(importer_type: type, namespace_handler: _NSHandlerType) -> None: ...
class IResourceProvider(IMetadataProvider): ...
class NullProvider: ...
class EggProvider(NullProvider): ...
class DefaultProvider(EggProvider): ...
class PathMetadata(DefaultProvider, IResourceProvider):
def __init__(self, path: str, egg_info: str) -> None: ...
class ZipProvider(EggProvider): ...
class EggMetadata(ZipProvider, IResourceProvider):
def __init__(self, zipimporter: zipimport.zipimporter) -> None: ...
class EmptyProvider(NullProvider): ...
empty_provider: EmptyProvider
class FileMetadata(EmptyProvider, IResourceProvider):
def __init__(self, path_to_pkg_info: str) -> None: ...
def parse_version(v: str) -> Tuple[str, ...]: ...
def yield_lines(strs: _NestedStr) -> Generator[str, None, None]: ...
def split_sections(strs: _NestedStr) -> Generator[Tuple[Optional[str], str], None, None]: ...
def safe_name(name: str) -> str: ...
def safe_version(version: str) -> str: ...
def safe_extra(extra: str) -> str: ...
def to_filename(name_or_version: str) -> str: ...
def get_build_platform() -> str: ...
def get_platform() -> str: ...
def get_supported_platform() -> str: ...
def compatible_platforms(provided: Optional[str], required: Optional[str]) -> bool: ...
def get_default_cache() -> str: ...
def get_importer(path_item: str) -> _Importer: ...
def ensure_directory(path: str) -> None: ...
def normalize_path(filename: str) -> str: ...
@@ -0,0 +1,7 @@
from typing import Text
import os
import sys
needs_makedirs: bool
makedirs = os.makedirs
@@ -0,0 +1,6 @@
from typing import Any, Tuple
from waitress.server import create_server
def serve(app: Any, **kw: Any) -> None: ...
def serve_paste(app: Any, global_conf: Any, **kw: Any) -> int: ...
def profile(cmd: Any, globals: Any, locals: Any, sort_order: Tuple[str, ...], callers: bool) -> None: ...
@@ -0,0 +1,60 @@
from .compat import HAS_IPV6, PY2, WIN, string_types
from .proxy_headers import PROXY_HEADERS
from socket import SocketType
from typing import Any, Dict, FrozenSet, Iterable, List, Sequence, Optional, Set, Tuple, Union
truthy: FrozenSet
KNOWN_PROXY_HEADERS: FrozenSet
def asbool(s: Optional[Union[bool, str, int]]) -> bool: ...
def asoctal(s: str) -> int: ...
def aslist_cronly(value: str) -> List[str]: ...
def aslist(value: str) -> List[str]: ...
def asset(value: Optional[str]) -> Set[str]: ...
def slash_fixed_str(s: Optional[str]) -> str: ...
def str_iftruthy(s: Optional[str]) -> Optional[str]: ...
def as_socket_list(sockets: Sequence[object]) -> List[SocketType]: ...
class _str_marker(str): ...
class _int_marker(int): ...
class _bool_marker: ...
class Adjustments:
host: _str_marker = ...
port: _int_marker = ...
listen: List[str] = ...
threads: int = ...
trusted_proxy: Optional[str] = ...
trusted_proxy_count: Optional[int] = ...
trusted_proxy_headers: Set[str] = ...
log_untrusted_proxy_headers: bool = ...
clear_untrusted_proxy_headers: Union[_bool_marker, bool] = ...
url_scheme: str = ...
url_prefix: str = ...
ident: str = ...
backlog: int = ...
recv_bytes: int = ...
send_bytes: int = ...
outbuf_overflow: int = ...
outbuf_high_watermark: int = ...
inbuf_overflow: int = ...
connection_limit: int = ...
cleanup_interval: int = ...
channel_timeout: int = ...
log_socket_errors: bool = ...
max_request_header_size: int = ...
max_request_body_size: int = ...
expose_tracebacks: bool = ...
unix_socket: Optional[str] = ...
unix_socket_perms: int = ...
socket_options: List[Tuple[int, int, int]] = ...
asyncore_loop_timeout: int = ...
asyncore_use_poll: bool = ...
ipv4: bool = ...
ipv6: bool = ...
sockets: List[SocketType] = ...
def __init__(self, **kw: Any) -> None: ...
@classmethod
def parse_args(cls, argv: str) -> Tuple[Dict[str, Any], Any]: ...
@classmethod
def check_sockets(cls, sockets: Iterable[SocketType]) -> None: ...
@@ -0,0 +1,57 @@
from io import BufferedIOBase, BufferedRandom, BytesIO
from typing import Any, Callable, Optional
COPY_BYTES: int
STRBUF_LIMIT: int
class FileBasedBuffer:
remain: int = ...
file: BytesIO = ...
def __init__(self, file: BytesIO, from_buffer: Optional[BytesIO] = ...) -> None: ...
def __len__(self) -> int: ...
def __nonzero__(self) -> bool: ...
__bool__: Callable[[], bool] = ...
def append(self, s: Any) -> None: ...
def get(self, numbytes: int = ..., skip: bool = ...) -> bytes: ...
def skip(self, numbytes: int, allow_prune: int = ...) -> None: ...
def newfile(self) -> Any: ...
def prune(self) -> None: ...
def getfile(self) -> Any: ...
def close(self) -> None: ...
class TempfileBasedBuffer(FileBasedBuffer):
def __init__(self, from_buffer: Optional[BytesIO] = ...) -> None: ...
def newfile(self) -> BufferedRandom: ...
class BytesIOBasedBuffer(FileBasedBuffer):
file: BytesIO = ...
def __init__(self, from_buffer: Optional[BytesIO] = ...) -> None: ...
def newfile(self) -> BytesIO: ...
class ReadOnlyFileBasedBuffer(FileBasedBuffer):
file: BytesIO = ...
block_size: int = ...
def __init__(self, file: BytesIO, block_size: int = ...) -> None: ...
remain: int = ...
def prepare(self, size: Optional[int] = ...) -> int: ...
def get(self, numbytes: int = ..., skip: bool = ...) -> bytes: ...
def __iter__(self) -> ReadOnlyFileBasedBuffer: ...
def next(self) -> Optional[bytes]: ...
__next__: Callable[[], Optional[bytes]] = ...
def append(self, s: Any) -> None: ...
class OverflowableBuffer:
overflowed: bool = ...
buf: Optional[BufferedIOBase] = ...
strbuf: bytes = ...
overflow: int = ...
def __init__(self, overflow: int) -> None: ...
def __len__(self) -> int: ...
def __nonzero__(self) -> bool: ...
__bool__: Callable[[], bool] = ...
def append(self, s: bytes) -> None: ...
def get(self, numbytes: int = ..., skip: bool = ...) -> bytes: ...
def skip(self, numbytes: int, allow_prune: bool = ...) -> None: ...
def prune(self) -> None: ...
def getfile(self) -> BytesIO: ...
def close(self) -> None: ...
@@ -0,0 +1,48 @@
from . import wasyncore as wasyncore
from socket import SocketType
from threading import Condition, Lock
from typing import Mapping, Sequence, Optional, Tuple
from waitress.adjustments import Adjustments
from waitress.buffers import OverflowableBuffer, ReadOnlyFileBasedBuffer
from waitress.parser import HTTPRequestParser
from waitress.server import BaseWSGIServer
from waitress.task import ErrorTask, WSGITask, Task
from waitress.utilities import InternalServerError
class ClientDisconnected(Exception): ...
class HTTPChannel(wasyncore.dispatcher):
task_class: WSGITask = ...
error_task_class: ErrorTask = ...
parser_class: HTTPRequestParser = ...
request: HTTPRequestParser = ...
last_activity: float = ...
will_close: bool = ...
close_when_flushed: bool = ...
requests: Sequence[HTTPRequestParser] = ...
sent_continue: bool = ...
total_outbufs_len: int = ...
current_outbuf_count: int = ...
server: BaseWSGIServer = ...
adj: Adjustments = ...
outbufs: Sequence[OverflowableBuffer] = ...
creation_time: float = ...
sendbuf_len: int = ...
task_lock: Lock = ...
outbuf_lock: Condition = ...
addr: Tuple[str, int] = ...
def __init__(
self, server: BaseWSGIServer, sock: SocketType, addr: str, adj: Adjustments, map: Optional[Mapping[int, SocketType]] = ...
) -> None: ...
def writable(self) -> bool: ...
def handle_write(self) -> None: ...
def readable(self) -> bool: ...
def handle_read(self) -> None: ...
def received(self, data: bytes) -> bool: ...
connected: bool = ...
def handle_close(self) -> None: ...
def add_channel(self, map: Optional[Mapping[int, SocketType]] = ...) -> None: ...
def del_channel(self, map: Optional[Mapping[int, SocketType]] = ...) -> None: ...
def write_soon(self, data: bytes) -> int: ...
def service(self) -> None: ...
def cancel(self) -> None: ...
@@ -0,0 +1,38 @@
from io import TextIOWrapper
import sys
from typing import Any, Optional, Text, Tuple
if sys.version_info[0] == 3:
from urllib import parse as urlparse
else:
import urlparse
PY2: bool
PY3: bool
WIN: bool
string_types: Tuple[str, ]
integer_types: Tuple[int, ]
class_types: Tuple[type, ]
text_type = str
binary_type = bytes
long = int
def unquote_bytes_to_wsgi(bytestring: bytes) -> str: ...
def text_(s: Text, encoding: str = ..., errors: str = ...) -> str: ...
def tostr(s: Text) -> str: ...
def tobytes(s: Text) -> bytes: ...
exec_: Any
def reraise(tp: Any, value: BaseException, tb: Optional[str] = ...) -> None: ...
MAXINT: int
HAS_IPV6: bool
IPPROTO_IPV6: int
IPV6_V6ONLY: int
def set_nonblocking(fd: TextIOWrapper) -> None: ...
ResourceWarning: Warning
def qualname(cls: Any) -> str: ...
@@ -0,0 +1,50 @@
from .rfc7230 import HEADER_FIELD
from io import BytesIO
from typing import Mapping, Sequence, Optional, Pattern, Tuple, Union
from waitress.adjustments import Adjustments
from waitress.buffers import OverflowableBuffer
from waitress.compat import tostr, unquote_bytes_to_wsgi, urlparse
from waitress.receiver import ChunkedReceiver, FixedStreamReceiver
from waitress.utilities import (
BadRequest,
RequestEntityTooLarge,
RequestHeaderFieldsTooLarge,
ServerNotImplemented,
find_double_newline,
Error,
)
class ParsingError(Exception): ...
class TransferEncodingNotImplemented(Exception): ...
class HTTPRequestParser:
completed: bool = ...
empty: bool = ...
expect_continue: bool = ...
headers_finished: bool = ...
header_plus: bytes = ...
chunked: bool = ...
content_length: int = ...
header_bytes_received: int = ...
body_bytes_received: int = ...
body_rcv: Optional[Union[ChunkedReceiver, FixedStreamReceiver]] = ...
version: str = ...
error: Optional[Error] = ...
connection_close: bool = ...
headers: Mapping[str, str] = ...
adj: Adjustments = ...
def __init__(self, adj: Adjustments) -> None: ...
def received(self, data: bytes) -> int: ...
first_line: str = ...
command: bytes = ...
url_scheme: str = ...
def parse_header(self, header_plus: bytes) -> None: ...
def get_body_stream(self) -> BytesIO: ...
def close(self) -> None: ...
def split_uri(uri: bytes) -> Tuple[str, str, bytes, str, str]: ...
def get_header_lines(header: bytes) -> Sequence[bytes]: ...
first_line_re: Pattern
def crack_first_line(line: str) -> Tuple[bytes, bytes, bytes]: ...
@@ -0,0 +1,31 @@
from .utilities import BadRequest, logger, undquote
from collections import namedtuple
from logging import Logger
from typing import Any, Callable, Mapping, Sequence, Optional, Set
PROXY_HEADERS: frozenset
Forwarded = namedtuple("Forwarded", ["by", "for_", "host", "proto"])
class MalformedProxyHeader(Exception):
header: str = ...
reason: str = ...
value: str = ...
def __init__(self, header: str, reason: str, value: str) -> None: ...
def proxy_headers_middleware(
app: Any,
trusted_proxy: Optional[str] = ...,
trusted_proxy_count: int = ...,
trusted_proxy_headers: Optional[Set[str]] = ...,
clear_untrusted: bool = ...,
log_untrusted: bool = ...,
logger: Logger = ...,
) -> Callable[..., Any]: ...
def parse_proxy_headers(
environ: Mapping[str, str], trusted_proxy_count: int, trusted_proxy_headers: Set[str], logger: Logger = ...
) -> Set[str]: ...
def strip_brackets(addr: str) -> str: ...
def clear_untrusted_headers(
environ: Mapping[str, str], untrusted_headers: Sequence[str], log_warning: bool = ..., logger: Logger = ...
) -> None: ...
@@ -0,0 +1,31 @@
from io import BytesIO
from typing import Optional
from waitress.buffers import OverflowableBuffer
from waitress.utilities import BadRequest, find_double_newline
class FixedStreamReceiver:
completed: bool = ...
error: None = ...
remain: int = ...
buf: OverflowableBuffer = ...
def __init__(self, cl: int, buf: OverflowableBuffer) -> None: ...
def __len__(self) -> int: ...
def received(self, data: bytes) -> int: ...
def getfile(self) -> BytesIO: ...
def getbuf(self) -> OverflowableBuffer: ...
class ChunkedReceiver:
chunk_remainder: int = ...
validate_chunk_end: bool = ...
control_line: bytes = ...
chunk_end: bytes = ...
all_chunks_received: bool = ...
trailer: bytes = ...
completed: bool = ...
error: Optional[BadRequest] = ...
buf: OverflowableBuffer = ...
def __init__(self, buf: OverflowableBuffer) -> None: ...
def __len__(self) -> int: ...
def received(self, s: bytes) -> int: ...
def getfile(self) -> BytesIO: ...
def getbuf(self) -> OverflowableBuffer: ...
@@ -0,0 +1,15 @@
from .compat import tobytes
from typing import Pattern
WS: str
OWS: str
RWS: str
BWS = str
TCHAR: str
OBS_TEXT: str
TOKEN: str
VCHAR: str
FIELD_VCHAR: str
FIELD_CONTENT: str
FIELD_VALUE: str
HEADER_FIELD: Pattern
@@ -0,0 +1,12 @@
from io import TextIOWrapper
from typing import Any, Callable, Sequence, Optional, Pattern, Tuple
from waitress import serve
HELP: str
RUNNER_PATTERN: Pattern
def match(obj_name: str) -> Tuple[str, str]: ...
def resolve(module_name: str, object_name: str) -> Any: ...
def show_help(stream: TextIOWrapper, name: str, error: Optional[str] = ...) -> None: ...
def show_exception(stream: TextIOWrapper) -> None: ...
def run(argv: Sequence[str] = ..., _serve: Callable[..., Any] = ...): ...
@@ -0,0 +1,105 @@
from . import wasyncore
from .proxy_headers import proxy_headers_middleware
from socket import SocketType
from typing import Any, Sequence, Optional, Tuple, Union
from waitress import trigger
from waitress.adjustments import Adjustments
from waitress.channel import HTTPChannel
from waitress.compat import IPPROTO_IPV6, IPV6_V6ONLY
from waitress.task import Task, ThreadedTaskDispatcher
from waitress.utilities import cleanup_unix_socket
def create_server(
application: Any,
map: Optional[Any] = ...,
_start: bool = ...,
_sock: Optional[SocketType] = ...,
_dispatcher: Optional[ThreadedTaskDispatcher] = ...,
**kw: Any,
) -> Union[MultiSocketServer, BaseWSGIServer]: ...
class MultiSocketServer:
asyncore: Any = ...
adj: Adjustments = ...
map: Any = ...
effective_listen: Sequence[Tuple[str, int]] = ...
task_dispatcher: ThreadedTaskDispatcher = ...
def __init__(
self,
map: Optional[Any] = ...,
adj: Optional[Adjustments] = ...,
effective_listen: Optional[Sequence[Tuple[str, int]]] = ...,
dispatcher: Optional[ThreadedTaskDispatcher] = ...,
) -> None: ...
def print_listen(self, format_str: str) -> None: ...
def run(self) -> None: ...
def close(self) -> None: ...
class BaseWSGIServer(wasyncore.dispatcher):
channel_class: HTTPChannel = ...
next_channel_cleanup: int = ...
socketmod: SocketType = ...
asyncore: Any = ...
sockinfo: Tuple[int, int, int, Tuple[str, int]] = ...
family: int = ...
socktype: int = ...
application: Any = ...
adj: Adjustments = ...
trigger: int = ...
task_dispatcher: ThreadedTaskDispatcher = ...
server_name: str = ...
active_channels: HTTPChannel = ...
def __init__(
self,
application: Any,
map: Optional[Any] = ...,
_start: bool = ...,
_sock: Optional[Any] = ...,
dispatcher: Optional[ThreadedTaskDispatcher] = ...,
adj: Optional[Adjustments] = ...,
sockinfo: Optional[Any] = ...,
bind_socket: bool = ...,
**kw: Any,
) -> None: ...
def bind_server_socket(self) -> None: ...
def get_server_name(self, ip: str) -> str: ...
def getsockname(self) -> Any: ...
accepting: bool = ...
def accept_connections(self) -> None: ...
def add_task(self, task: Task) -> None: ...
def readable(self) -> bool: ...
def writable(self) -> bool: ...
def handle_read(self) -> None: ...
def handle_connect(self) -> None: ...
def handle_accept(self) -> None: ...
def run(self) -> None: ...
def pull_trigger(self) -> None: ...
def set_socket_options(self, conn: Any) -> None: ...
def fix_addr(self, addr: Any) -> Any: ...
def maintenance(self, now: int) -> None: ...
def print_listen(self, format_str: str) -> None: ...
def close(self) -> None: ...
class TcpWSGIServer(BaseWSGIServer):
def bind_server_socket(self) -> None: ...
def getsockname(self) -> Tuple[str, Tuple[str, int]]: ...
def set_socket_options(self, conn: SocketType) -> None: ...
class UnixWSGIServer(BaseWSGIServer):
def __init__(
self,
application: Any,
map: Optional[Any] = ...,
_start: bool = ...,
_sock: Optional[Any] = ...,
dispatcher: Optional[Any] = ...,
adj: Optional[Adjustments] = ...,
sockinfo: Optional[Any] = ...,
**kw: Any,
) -> None: ...
def bind_server_socket(self) -> None: ...
def getsockname(self) -> Tuple[str, Tuple[str, int]]: ...
def fix_addr(self, addr: Any) -> Tuple[str, None]: ...
def get_server_name(self, ip: Any) -> str: ...
WSGIServer: TcpWSGIServer
+70
View File
@@ -0,0 +1,70 @@
from .buffers import ReadOnlyFileBasedBuffer
from .channel import HTTPChannel
from .compat import reraise, tobytes
from .utilities import Error, build_http_date, logger, queue_logger
from logging import Logger
from threading import Condition, Lock
from typing import Any, Deque, Mapping, Sequence, Optional, Set, Tuple
rename_headers: Mapping[str, str]
hop_by_hop: frozenset
class ThreadedTaskDispatcher:
stop_count: int = ...
active_count: int = ...
logger: Logger = ...
queue_logger: Logger = ...
threads: Set = ...
queue: Deque[Task] = ...
lock: Lock = ...
queue_cv: Condition = ...
thread_exit_cv: Condition = ...
def __init__(self) -> None: ...
def start_new_thread(self, target: Any, args: Any) -> None: ...
def handler_thread(self, thread_no: int) -> None: ...
def set_thread_count(self, count: int) -> None: ...
def add_task(self, task: Task) -> None: ...
def shutdown(self, cancel_pending: bool = ..., timeout: int = ...) -> bool: ...
class Task:
close_on_finish: bool = ...
status: str = ...
wrote_header: bool = ...
start_time: int = ...
content_length: Optional[int] = ...
content_bytes_written: int = ...
logged_write_excess: bool = ...
logged_write_no_body: bool = ...
complete: bool = ...
chunked_response: bool = ...
logger: Logger = ...
channel: HTTPChannel = ...
request: Error = ...
response_headers: Sequence[Tuple[str, str]] = ...
version: str = ...
def __init__(self, channel: HTTPChannel, request: Error) -> None: ...
def service(self) -> None: ...
@property
def has_body(self) -> bool: ...
def build_response_header(self) -> bytes: ...
def remove_content_length_header(self) -> None: ...
def start(self) -> None: ...
def finish(self) -> None: ...
def write(self, data: bytes) -> None: ...
class ErrorTask(Task):
complete: bool = ...
status: str = ...
close_on_finish: bool = ...
content_length: int = ...
def execute(self) -> None: ...
class WSGITask(Task):
environ: Optional[Any] = ...
response_headers: Sequence[Tuple[str, str]] = ...
complete: bool = ...
status: str = ...
content_length: int = ...
close_on_finish: bool = ...
def execute(self) -> None: ...
def get_environment(self) -> Any: ...
@@ -0,0 +1,30 @@
from . import wasyncore as wasyncore
from threading import Lock
from socket import SocketType
import sys
from typing import Callable, Mapping, Optional
from typing_extensions import Literal
class _triggerbase:
kind: Optional[str] = ...
lock: Lock = ...
thunks: Callable[[None], None] = ...
def __init__(self) -> None: ...
def readable(self) -> Literal[True]: ...
def writable(self) -> Literal[False]: ...
def handle_connect(self) -> None: ...
def handle_close(self) -> None: ...
def close(self) -> None: ...
def pull_trigger(self, thunk: Optional[Callable[[None], None]] = ...) -> None: ...
def handle_read(self) -> None: ...
if sys.platform == "linux" or sys.platform == "darwin":
class trigger(_triggerbase, wasyncore.file_dispatcher):
kind: str = ...
def __init__(self, map: Mapping[str, _triggerbase]) -> None: ...
else:
class trigger(_triggerbase, wasyncore.dispatcher):
kind: str = ...
trigger: SocketType = ...
def __init__(self, map: Mapping[str, _triggerbase]) -> None: ...
@@ -0,0 +1,75 @@
from .rfc7230 import OBS_TEXT, VCHAR
from logging import Logger
from typing import Any, Callable, Mapping, Sequence, Match, Pattern, Tuple
logger: Logger
queue_logger: Logger
def find_double_newline(s: bytes) -> int: ...
def concat(*args: Any) -> str: ...
def join(seq: Any, field: str = ...) -> str: ...
def group(s: Any) -> str: ...
short_days: Sequence[str]
long_days: Sequence[str]
short_day_reg: str
long_day_reg: str
daymap: Mapping[str, int]
hms_reg: str
months: Sequence[str]
monmap: Mapping[str, int]
months_reg: str
rfc822_date: str
rfc822_reg: Pattern
def unpack_rfc822(m: Match) -> Tuple[int, int, int, int, int, int, int, int, int]: ...
rfc850_date: str
rfc850_reg: Pattern
def unpack_rfc850(m: Match) -> Tuple[int, int, int, int, int, int, int, int, int]: ...
weekdayname: Sequence[str]
monthname: Sequence[str]
def build_http_date(when: int) -> str: ...
def parse_http_date(d: str) -> int: ...
vchar_re: str
obs_text_re: str
qdtext_re: str
quoted_pair_re: str
quoted_string_re: str
quoted_string: Pattern
quoted_pair: Pattern
def undquote(value: str) -> str: ...
def cleanup_unix_socket(path: str) -> None: ...
class Error:
code: int = ...
reason: str = ...
body: str = ...
def __init__(self, body: str) -> None: ...
def to_response(self) -> Tuple[str, Sequence[Tuple[str, str]], str]: ...
def wsgi_response(self, environ: Any, start_response: Callable[[str, Sequence[Tuple[str, str]]], None]) -> str: ...
class BadRequest(Error):
code: int = ...
reason: str = ...
class RequestHeaderFieldsTooLarge(BadRequest):
code: int = ...
reason: str = ...
class RequestEntityTooLarge(BadRequest):
code: int = ...
reason: str = ...
class InternalServerError(Error):
code: int = ...
reason: str = ...
class ServerNotImplemented(Error):
code: int = ...
reason: str = ...
@@ -0,0 +1,93 @@
from . import compat, utilities
from io import BytesIO
from logging import Logger
from socket import SocketType
from typing import Any, Callable, Mapping, Optional, Tuple
socket_map: Mapping[int, SocketType]
map: Mapping[int, SocketType]
class ExitNow(Exception): ...
def read(obj: dispatcher) -> None: ...
def write(obj: dispatcher) -> None: ...
def readwrite(obj: dispatcher, flags: int) -> None: ...
def poll(timeout: float = ..., map: Optional[Mapping[int, SocketType]] = ...) -> None: ...
def poll2(timeout: float = ..., map: Optional[Mapping[int, SocketType]] = ...) -> None: ...
poll3 = poll2
def loop(
timeout: float = ..., use_poll: bool = ..., map: Optional[Mapping[int, SocketType]] = ..., count: Optional[int] = ...
) -> None: ...
def compact_traceback() -> Tuple[Tuple[str, str, str], BaseException, BaseException, str]: ...
class dispatcher:
debug: bool = ...
connected: bool = ...
accepting: bool = ...
connecting: bool = ...
closing: bool = ...
addr: Optional[Tuple[str, int]] = ...
ignore_log_types: frozenset = ...
logger: Logger = ...
compact_traceback: Callable[[], Tuple[Tuple[str, str, str], BaseException, BaseException, str]] = ...
socket: Optional[SocketType] = ...
def __init__(self, sock: Optional[SocketType] = ..., map: Optional[Mapping[int, SocketType]] = ...) -> None: ...
def add_channel(self, map: Optional[Mapping[int, SocketType]] = ...) -> None: ...
def del_channel(self, map: Optional[Mapping[int, SocketType]] = ...) -> None: ...
family_and_type: Tuple[int, int] = ...
def create_socket(self, family: int = ..., type: int = ...) -> None: ...
def set_socket(self, sock: SocketType, map: Optional[Mapping[int, SocketType]] = ...) -> None: ...
def set_reuse_addr(self) -> None: ...
def readable(self) -> bool: ...
def writable(self) -> bool: ...
def listen(self, num: int) -> None: ...
def bind(self, addr: Tuple[str, int]) -> None: ...
def connect(self, address: Tuple[str, int]) -> None: ...
def accept(self) -> Optional[Tuple[SocketType, Tuple[str, int]]]: ...
def send(self, data: bytes) -> int: ...
def recv(self, buffer_size: int) -> bytes: ...
def close(self) -> None: ...
def log(self, message: str) -> None: ...
def log_info(self, message: str, type: str = ...) -> None: ...
def handle_read_event(self) -> None: ...
def handle_connect_event(self) -> None: ...
def handle_write_event(self) -> None: ...
def handle_expt_event(self) -> None: ...
def handle_error(self) -> None: ...
def handle_expt(self) -> None: ...
def handle_read(self) -> None: ...
def handle_write(self) -> None: ...
def handle_connect(self) -> None: ...
def handle_accept(self) -> None: ...
def handle_accepted(self, sock: SocketType, addr: Any) -> None: ...
def handle_close(self) -> None: ...
class dispatcher_with_send(dispatcher):
out_buffer: bytes = ...
def __init__(self, sock: Optional[SocketType] = ..., map: Optional[Mapping[int, SocketType]] = ...) -> None: ...
def initiate_send(self) -> None: ...
handle_write: Callable[[], None] = ...
def writable(self) -> bool: ...
def send(self, data: bytes) -> None: ... # type: ignore
def close_all(map: Optional[Mapping[int, SocketType]] = ..., ignore_all: bool = ...) -> None: ...
class file_wrapper:
fd: BytesIO = ...
def __init__(self, fd: BytesIO) -> None: ...
def __del__(self) -> None: ...
def recv(self, *args: Any) -> bytes: ...
def send(self, *args: Any) -> bytes: ...
def getsockopt(self, level: int, optname: int, buflen: Optional[bool] = ...) -> int: ...
read: Callable[..., bytes] = ...
write: Callable[..., bytes] = ...
def close(self) -> None: ...
def fileno(self) -> BytesIO: ...
class file_dispatcher(dispatcher):
connected: bool = ...
def __init__(self, fd: BytesIO, map: Optional[Mapping[int, SocketType]] = ...) -> None: ...
socket: SocketType = ...
def set_file(self, fd: BytesIO) -> None: ...
@@ -28,7 +28,10 @@ val whiteList = sequenceOf(
"__builtin__",
"__future__",
//"_ast", leads to broken tests but could be enabled
"_bisect",
"_bootlocale",
"_codecs",
"_compat_pickle",
"_compression",
"_csv",
"_curses",
@@ -38,28 +41,40 @@ val whiteList = sequenceOf(
"_importlib_modulespec",
"_io",
"_json",
"_msi",
"_operator",
"_py_abc",
"_random",
"_sitebuiltins",
"_thread",
"_tracemalloc",
"_types",
"_warnings",
"_winapi",
"_winreg",
"abc",
"aifc",
"antigravity",
"argparse",
"array",
"ast",
"asyncio",
"asyncore",
"atomicwrites",
"attr",
"audioop",
"base64",
"bdb",
"binascii",
"builtins",
"bz2",
"cgi",
"chardet",
"cmath",
"cmd",
"codecs",
"collections",
"compileall",
"concurrent",
"configparser",
"contextvars",
@@ -84,6 +99,7 @@ val whiteList = sequenceOf(
"fcntl",
"filecmp",
//"formatter", leads to broken tests but could be enabled
"ftplib",
"functools",
"gc",
"genericpath",
@@ -91,6 +107,7 @@ val whiteList = sequenceOf(
"gflags",
"hashlib",
"heapq",
"hmac",
"http",
"imaplib",
"imghdr",
@@ -102,12 +119,15 @@ val whiteList = sequenceOf(
"json",
"locale",
"logging",
"lzma",
"macpath",
"marshal",
"math",
"mimetypes",
"mmap",
"mock",
"modulefinder",
"msilib",
"multiprocessing",
"ntpath",
"numbers",
@@ -122,6 +142,7 @@ val whiteList = sequenceOf(
"pathlib2",
"pdb",
"pickle",
"pkg_resources",
//"platform", leads to broken tests but could be enabled
"plistlib",
"posix",
@@ -132,6 +153,7 @@ val whiteList = sequenceOf(
"pydoc",
"pyexpat",
"queue",
"random",
"re",
"requests",
"resource",
@@ -150,6 +172,7 @@ val whiteList = sequenceOf(
//"string", leads to broken tests but could be enabled
"struct",
"subprocess",
"sunau",
"sys",
"tarfile",
"tempfile",
@@ -160,20 +183,28 @@ val whiteList = sequenceOf(
"time",
"token",
"tokenize",
"tracemalloc",
"turtle",
"types",
"typing",
"typing_extensions",
"tzlocal",
"unittest",
"urllib",
"urlparse",
"uu",
"uuid",
"venv",
"waitress",
"warnings",
"wave",
"webbrowser",
"werkzeug",
"winreg",
"winsound",
"wsgiref",
"xml",
"xmlrpc",
"zipapp",
"zipfile",
"zipimport",