Add type stub for PySocks (#14623)

This commit is contained in:
Chocobo1
2025-08-27 21:19:46 +08:00
committed by GitHub
parent dd31a8a879
commit 61107f2860
4 changed files with 250 additions and 0 deletions
@@ -0,0 +1,3 @@
# Internal variables that were improperly leaked to the outside
socks.method
socks.name
+2
View File
@@ -0,0 +1,2 @@
version = "1.7.1"
upstream_repository = "https://github.com/Anorov/PySocks"
+143
View File
@@ -0,0 +1,143 @@
import logging
import socket
import types
from _typeshed import ReadableBuffer
from collections.abc import Callable, Iterable, Mapping
from typing import Final, TypeVar, overload
from typing_extensions import ParamSpec, TypeAlias
__version__: Final[str]
log: logging.Logger # undocumented
_ProxyType: TypeAlias = int
PROXY_TYPE_SOCKS4: Final[_ProxyType]
SOCKS4: Final[_ProxyType]
PROXY_TYPE_SOCKS5: Final[_ProxyType]
SOCKS5: Final[_ProxyType]
PROXY_TYPE_HTTP: Final[_ProxyType]
HTTP: Final[_ProxyType]
PROXY_TYPES: Final[dict[str, _ProxyType]]
PRINTABLE_PROXY_TYPES: Final[dict[_ProxyType, str]]
_T = TypeVar("_T")
_P = ParamSpec("_P")
def set_self_blocking(function: Callable[_P, _T]) -> Callable[_P, _T]: ... # undocumented
class ProxyError(IOError):
msg: str
socket_err: socket.error
def __init__(self, msg: str, socket_err: socket.error | None = None) -> None: ...
class GeneralProxyError(ProxyError): ...
class ProxyConnectionError(ProxyError): ...
class SOCKS5AuthError(ProxyError): ...
class SOCKS5Error(ProxyError): ...
class SOCKS4Error(ProxyError): ...
class HTTPError(ProxyError): ...
SOCKS4_ERRORS: Final[Mapping[int, str]]
SOCKS5_ERRORS: Final[Mapping[int, str]]
DEFAULT_PORTS: Final[Mapping[_ProxyType, int]]
_DefaultProxy: TypeAlias = tuple[_ProxyType | None, str | None, int | None, bool, bytes | None, bytes | None]
def set_default_proxy(
proxy_type: _ProxyType | None = None,
addr: str | None = None,
port: int | None = None,
rdns: bool = True,
username: str | None = None,
password: str | None = None,
) -> None: ...
def setdefaultproxy(
proxy_type: _ProxyType | None = None,
addr: str | None = None,
port: int | None = None,
rdns: bool = True,
username: str | None = None,
password: str | None = None,
*,
proxytype: _ProxyType = ...,
) -> None: ...
def get_default_proxy() -> _DefaultProxy | None: ...
getdefaultproxy = get_default_proxy
def wrap_module(module: types.ModuleType) -> None: ...
wrapmodule = wrap_module
_Endpoint: TypeAlias = tuple[str, int]
def create_connection(
dest_pair: _Endpoint,
timeout: int | None = None,
source_address: _Endpoint | None = None,
proxy_type: _ProxyType | None = None,
proxy_addr: str | None = None,
proxy_port: int | None = None,
proxy_rdns: bool = True,
proxy_username: str | None = None,
proxy_password: str | None = None,
socket_options: (
Iterable[tuple[int, int, int | ReadableBuffer] | tuple[int, int, None, int]] | None
) = None, # values passing to `socket.setsockopt` method
) -> socksocket: ...
class _BaseSocket(socket.socket): # undocumented
...
class socksocket(_BaseSocket):
default_proxy: _DefaultProxy | None # undocumented
proxy: _DefaultProxy # undocumented
proxy_sockname: _Endpoint | None # undocumented
proxy_peername: _Endpoint | None # undocumented
def __init__(
self, family: socket.AddressFamily = ..., type: socket.SocketKind = ..., proto: int = 0, fileno: int | None = None
) -> None: ...
def settimeout(self, timeout: float | None) -> None: ...
def gettimeout(self) -> float | None: ...
def setblocking(self, v: bool) -> None: ...
def set_proxy(
self,
proxy_type: _ProxyType | None = None,
addr: str | None = None,
port: int | None = None,
rdns: bool = True,
username: str | None = None,
password: str | None = None,
) -> None: ...
def setproxy(
self,
proxy_type: _ProxyType | None = None,
addr: str | None = None,
port: int | None = None,
rdns: bool = True,
username: str | None = None,
password: str | None = None,
*,
proxytype: _ProxyType = ...,
) -> None: ...
def bind(self, address: socket._Address, /) -> None: ...
@overload
def sendto(self, bytes: ReadableBuffer, address: socket._Address) -> int: ...
@overload
def sendto(self, bytes: ReadableBuffer, flags: int, address: socket._Address) -> int: ...
def send(self, bytes: ReadableBuffer, flags: int = 0) -> int: ...
def recvfrom(self, bufsize: int, flags: int = 0) -> tuple[bytes, _Endpoint]: ...
def recv(self, bufsize: int, flags: int = 0) -> bytes: ...
def close(self) -> None: ...
def get_proxy_sockname(self) -> _Endpoint | None: ...
getproxysockname = get_proxy_sockname
def get_proxy_peername(self) -> _Endpoint | None: ...
getproxypeername = get_proxy_peername
def get_peername(self) -> _Endpoint | None: ...
getpeername = get_peername
@set_self_blocking
def connect(self, dest_pair: _Endpoint, catch_errors: bool | None = None) -> None: ... # type: ignore[override]
@set_self_blocking
def connect_ex(self, dest_pair: _Endpoint) -> int: ... # type: ignore[override]
+102
View File
@@ -0,0 +1,102 @@
import http.client
import ssl
import sys
import urllib.request
from _typeshed import Incomplete, SupportsKeysAndGetItem
from typing import Any, TypeVar
from typing_extensions import override
import socks
_K = TypeVar("_K")
_V = TypeVar("_V")
def merge_dict(a: dict[_K, _V], b: SupportsKeysAndGetItem[_K, _V]) -> dict[_K, _V]: ... # undocumented
def is_ip(s: str) -> bool: ... # undocumented
socks4_no_rdns: set[str] # undocumented
class SocksiPyConnection(http.client.HTTPConnection): # undocumented
proxyargs: tuple[int, str, int | None, bool, str | None, str | None]
sock: socks.socksocket
def __init__(
self,
proxytype: int,
proxyaddr: str,
proxyport: int | None = None,
rdns: bool = True,
username: str | None = None,
password: str | None = None,
host: str | None = None,
port: int | None = None,
timeout: float | None = ...,
source_address: tuple[str, int] | None = None,
blocksize: int = 8192,
) -> None: ...
@override
def connect(self) -> None: ...
class SocksiPyConnectionS(http.client.HTTPSConnection): # undocumented
proxyargs: tuple[int, str, int | None, bool, str | None, str | None]
sock: socks.socksocket
if sys.version_info >= (3, 12):
def __init__(
self,
proxytype: int,
proxyaddr: str,
proxyport: int | None = None,
rdns: bool = True,
username: str | None = None,
password: str | None = None,
host: str | None = None,
port: int | None = None,
*,
timeout: float | None = ...,
source_address: tuple[str, int] | None = None,
context: ssl.SSLContext | None = None,
blocksize: int = 8192,
) -> None: ...
else:
def __init__(
self,
proxytype: int,
proxyaddr: str,
proxyport: int | None = None,
rdns: bool = True,
username: str | None = None,
password: str | None = None,
host: str | None = None,
port: int | None = None,
key_file: str | None = None,
cert_file: str | None = None,
timeout: float | None = ...,
source_address: tuple[str, int] | None = None,
*,
context: ssl.SSLContext | None = None,
check_hostname: bool | None = None,
blocksize: int = 8192,
) -> None: ...
@override
def connect(self) -> None: ...
class SocksiPyHandler(urllib.request.HTTPHandler, urllib.request.HTTPSHandler):
args: tuple[Incomplete, ...] # undocumented
kw: dict[str, Incomplete] # undocumented
def __init__(
self,
proxytype: int,
proxyaddr: str,
proxyport: int | None = None,
rdns: bool = True,
username: str | None = None,
password: str | None = None,
*,
source_address: tuple[str, int] | None = None,
blocksize: int = 8192,
**kwargs: Any, # any additional arguments to `SocksiPyConnection` or `SocksiPyConnectionS`
) -> None: ...
@override
def http_open(self, req: urllib.request.Request) -> http.client.HTTPResponse: ... # undocumented
@override
def https_open(self, req: urllib.request.Request) -> http.client.HTTPResponse: ... # undocumented