From 1128f598e4c48237e065c45c61dc916ce876409a Mon Sep 17 00:00:00 2001 From: donbarbos Date: Tue, 21 Jul 2026 19:54:08 +0400 Subject: [PATCH 1/2] [pika] Complete adapters.utils --- .../adapters/utils/connection_workflow.pyi | 42 +++++-- .../pika/adapters/utils/io_services_utils.pyi | 86 ++++++++++---- .../pika/adapters/utils/nbio_interface.pyi | 80 ++++++++----- .../utils/selector_ioloop_adapter.pyi | 105 ++++++++++++------ 4 files changed, 220 insertions(+), 93 deletions(-) diff --git a/stubs/pika/pika/adapters/utils/connection_workflow.pyi b/stubs/pika/pika/adapters/utils/connection_workflow.pyi index 1adf533f8ce9..a2284221bde8 100644 --- a/stubs/pika/pika/adapters/utils/connection_workflow.pyi +++ b/stubs/pika/pika/adapters/utils/connection_workflow.pyi @@ -1,6 +1,9 @@ from _typeshed import Incomplete +from collections.abc import Callable, Iterable, Sequence import pika.compat +import pika.connection +from pika.adapters.utils.nbio_interface import AbstractIOServices, AbstractStreamProtocol class AMQPConnectorException(Exception): ... class AMQPConnectorStackTimeout(AMQPConnectorException): ... @@ -8,8 +11,8 @@ class AMQPConnectorAborted(AMQPConnectorException): ... class AMQPConnectorWrongState(AMQPConnectorException): ... class AMQPConnectorPhaseErrorBase(AMQPConnectorException): - exception: Incomplete - def __init__(self, exception, *args) -> None: ... + exception: BaseException + def __init__(self, exception: BaseException, *args: object) -> None: ... class AMQPConnectorSocketConnectError(AMQPConnectorPhaseErrorBase): ... class AMQPConnectorTransportSetupError(AMQPConnectorPhaseErrorBase): ... @@ -18,20 +21,41 @@ class AMQPConnectionWorkflowAborted(AMQPConnectorException): ... class AMQPConnectionWorkflowWrongState(AMQPConnectorException): ... class AMQPConnectionWorkflowFailed(AMQPConnectorException): - exceptions: Incomplete - def __init__(self, exceptions, *args) -> None: ... + exceptions: tuple[BaseException, ...] + def __init__(self, exceptions: Iterable[BaseException], *args: object) -> None: ... class AMQPConnector: - def __init__(self, conn_factory, nbio) -> None: ... - def start(self, addr_record, conn_params, on_done) -> None: ... + def __init__( + self, conn_factory: Callable[[pika.connection.Parameters], AbstractStreamProtocol], nbio: AbstractIOServices + ) -> None: ... + def start( + self, + addr_record: tuple[ # tuple taken result of socket.getaddrinfo + int, int, int, str, tuple[str, int] | tuple[str, int, int, int] | tuple[int, bytes] + ], + conn_params: pika.connection.Parameters, + on_done: Callable[[pika.connection.Connection | BaseException], None], + ) -> None: ... def abort(self) -> None: ... class AbstractAMQPConnectionWorkflow(pika.compat.AbstractBase): - def start(self, connection_configs, connector_factory, native_loop, on_done) -> None: ... + def start( + self, + connection_configs: Sequence[pika.connection.Parameters], + connector_factory: Callable[..., Incomplete], + native_loop, + on_done: Callable[[pika.connection.Connection | AMQPConnectorException], None], + ) -> None: ... def abort(self) -> None: ... class AMQPConnectionWorkflow(AbstractAMQPConnectionWorkflow): def __init__(self, _until_first_amqp_attempt: bool = False) -> None: ... - def set_io_services(self, nbio) -> None: ... - def start(self, connection_configs, connector_factory, native_loop, on_done) -> None: ... + def set_io_services(self, nbio: AbstractIOServices) -> None: ... + def start( + self, + connection_configs: Sequence[pika.connection.Parameters], + connector_factory: Callable[..., Incomplete], + native_loop, + on_done: Callable[[pika.connection.Connection | AMQPConnectorException], None], + ) -> None: ... def abort(self) -> None: ... diff --git a/stubs/pika/pika/adapters/utils/io_services_utils.pyi b/stubs/pika/pika/adapters/utils/io_services_utils.pyi index 3f99e672ee31..604e533bc5f3 100644 --- a/stubs/pika/pika/adapters/utils/io_services_utils.pyi +++ b/stubs/pika/pika/adapters/utils/io_services_utils.pyi @@ -1,43 +1,89 @@ import abc +from collections.abc import Callable +from socket import socket +from ssl import SSLContext, SSLSocket +from typing import Any -from pika.adapters.utils.nbio_interface import AbstractIOReference, AbstractStreamTransport +from pika.adapters.utils.nbio_interface import ( + AbstractFileDescriptorServices, + AbstractIOReference, + AbstractIOServices, + AbstractStreamProtocol, + AbstractStreamTransport, +) +from pika.adapters.utils.selector_ioloop_adapter import _SupportsCancel -def check_callback_arg(callback, name) -> None: ... -def check_fd_arg(fd) -> None: ... +def check_callback_arg(callback: Callable[..., Any], name: str) -> None: ... +def check_fd_arg(fd: int) -> None: ... class SocketConnectionMixin: - def connect_socket(self, sock, resolved_addr, on_done): ... + def connect_socket( + self, sock: socket, resolved_addr: tuple[str, int], on_done: Callable[[BaseException | None], None] + ) -> _AsyncServiceAsyncHandle: ... class StreamingConnectionMixin: - def create_streaming_connection(self, protocol_factory, sock, on_done, ssl_context=None, server_hostname=None): ... + def create_streaming_connection( + self, + protocol_factory: Callable[[], AbstractStreamProtocol], + sock: socket, + on_done: Callable[[tuple[AbstractStreamTransport, AbstractStreamProtocol] | BaseException], None], + ssl_context: SSLContext | None = None, + server_hostname: str | None = None, + ) -> AbstractIOReference: ... class _AsyncServiceAsyncHandle(AbstractIOReference): - def __init__(self, subject) -> None: ... - def cancel(self): ... + def __init__(self, subject: _SupportsCancel) -> None: ... + def cancel(self) -> bool: ... class _AsyncSocketConnector: - def __init__(self, nbio, sock, resolved_addr, on_done) -> None: ... - def start(self): ... - def cancel(self): ... + def __init__( + self, + nbio: AbstractIOServices | AbstractFileDescriptorServices, + sock: socket, + resolved_addr: tuple[str, int], + on_done: Callable[[BaseException | None], None], + ) -> None: ... + def start(self) -> AbstractIOReference: ... + def cancel(self) -> bool: ... class _AsyncStreamConnector: - def __init__(self, nbio, protocol_factory, sock, ssl_context, server_hostname, on_done) -> None: ... - def start(self): ... - def cancel(self): ... + def __init__( + self, + nbio: AbstractIOServices | AbstractFileDescriptorServices, + protocol_factory: Callable[[], AbstractStreamProtocol], + sock: socket, + ssl_context: SSLContext, + server_hostname: str | None, + on_done: Callable[[tuple[AbstractStreamTransport, AbstractStreamProtocol] | BaseException], None], + ) -> None: ... + def start(self) -> AbstractIOReference: ... + def cancel(self) -> bool: ... class _AsyncTransportBase(AbstractStreamTransport, metaclass=abc.ABCMeta): class RxEndOfFile(OSError): def __init__(self) -> None: ... - def __init__(self, sock, protocol, nbio) -> None: ... + def __init__( + self, + sock: socket | SSLSocket, + protocol: AbstractStreamProtocol, + nbio: AbstractIOServices | AbstractFileDescriptorServices, + ) -> None: ... def abort(self) -> None: ... - def get_protocol(self): ... - def get_write_buffer_size(self): ... + def get_protocol(self) -> AbstractStreamProtocol: ... + def get_write_buffer_size(self) -> int: ... class _AsyncPlaintextTransport(_AsyncTransportBase): - def __init__(self, sock, protocol, nbio) -> None: ... - def write(self, data) -> None: ... + def __init__( + self, + sock: socket | SSLSocket, + protocol: AbstractStreamProtocol, + nbio: AbstractIOServices | AbstractFileDescriptorServices, + ) -> None: ... + def write(self, data: bytes) -> None: ... class _AsyncSSLTransport(_AsyncTransportBase): - def __init__(self, sock, protocol, nbio) -> None: ... - def write(self, data) -> None: ... + def __init__( + self, sock: SSLSocket, protocol: AbstractStreamProtocol, nbio: AbstractIOServices | AbstractFileDescriptorServices + ) -> None: ... + def write(self, data: bytes) -> None: ... diff --git a/stubs/pika/pika/adapters/utils/nbio_interface.pyi b/stubs/pika/pika/adapters/utils/nbio_interface.pyi index 14c721842d29..6b963642813e 100644 --- a/stubs/pika/pika/adapters/utils/nbio_interface.pyi +++ b/stubs/pika/pika/adapters/utils/nbio_interface.pyi @@ -1,61 +1,85 @@ import abc +from collections.abc import Callable +from socket import socket +from ssl import SSLContext import pika.compat class AbstractIOServices(pika.compat.AbstractBase, metaclass=abc.ABCMeta): @abc.abstractmethod - def get_native_ioloop(self): ... + def get_native_ioloop(self) -> object: ... @abc.abstractmethod - def close(self): ... + def close(self) -> None: ... @abc.abstractmethod - def run(self): ... + def run(self) -> None: ... @abc.abstractmethod - def stop(self): ... + def stop(self) -> None: ... @abc.abstractmethod - def add_callback_threadsafe(self, callback): ... + def add_callback_threadsafe(self, callback: Callable[..., None]) -> None: ... @abc.abstractmethod - def call_later(self, delay, callback): ... + def call_later(self, delay: float, callback: Callable[..., None]) -> AbstractTimerReference: ... @abc.abstractmethod - def getaddrinfo(self, host, port, on_done, family: int = 0, socktype: int = 0, proto: int = 0, flags: int = 0): ... + def getaddrinfo( + self, + host: str | bytes | None, + port: str | bytes | int | None, + on_done: Callable[ # list is result of socket.getaddrinfo + [list[tuple[int, int, int, str, tuple[str, int] | tuple[str, int, int, int] | tuple[int, bytes]]] | BaseException], + None, + ], + family: int = 0, + socktype: int = 0, + proto: int = 0, + flags: int = 0, + ) -> AbstractIOReference: ... @abc.abstractmethod - def connect_socket(self, sock, resolved_addr, on_done): ... + def connect_socket( + self, sock: socket, resolved_addr: tuple[str, int], on_done: Callable[[BaseException | None], None] + ) -> AbstractIOReference: ... @abc.abstractmethod - def create_streaming_connection(self, protocol_factory, sock, on_done, ssl_context=None, server_hostname=None): ... + def create_streaming_connection( + self, + protocol_factory: Callable[[], AbstractStreamProtocol], + sock: socket, + on_done: Callable[[tuple[AbstractStreamTransport, AbstractStreamProtocol] | BaseException], None], + ssl_context: SSLContext | None = None, + server_hostname: str | None = None, + ) -> AbstractIOReference: ... -class AbstractFileDescriptorServices(pika.compat.AbstractBase, metaclass=abc.ABCMeta): +class AbstractFileDescriptorServices(pika.compat.AbstractBase): @abc.abstractmethod - def set_reader(self, fd, on_readable): ... + def set_reader(self, fd: int, on_readable: Callable[[], None]) -> None: ... @abc.abstractmethod - def remove_reader(self, fd): ... + def remove_reader(self, fd: int) -> bool: ... @abc.abstractmethod - def set_writer(self, fd, on_writable): ... + def set_writer(self, fd: int, on_writable: Callable[[], None]) -> None: ... @abc.abstractmethod - def remove_writer(self, fd): ... + def remove_writer(self, fd: int) -> bool: ... -class AbstractTimerReference(pika.compat.AbstractBase, metaclass=abc.ABCMeta): +class AbstractTimerReference(pika.compat.AbstractBase): @abc.abstractmethod - def cancel(self): ... + def cancel(self) -> None: ... -class AbstractIOReference(pika.compat.AbstractBase, metaclass=abc.ABCMeta): +class AbstractIOReference(pika.compat.AbstractBase): @abc.abstractmethod - def cancel(self): ... + def cancel(self) -> bool: ... -class AbstractStreamProtocol(pika.compat.AbstractBase, metaclass=abc.ABCMeta): +class AbstractStreamProtocol(pika.compat.AbstractBase): @abc.abstractmethod - def connection_made(self, transport): ... + def connection_made(self, transport: AbstractStreamTransport) -> None: ... @abc.abstractmethod - def connection_lost(self, error): ... + def connection_lost(self, error: BaseException | None) -> None: ... @abc.abstractmethod - def eof_received(self): ... + def eof_received(self) -> bool | None: ... @abc.abstractmethod - def data_received(self, data): ... + def data_received(self, data: bytes) -> None: ... -class AbstractStreamTransport(pika.compat.AbstractBase, metaclass=abc.ABCMeta): +class AbstractStreamTransport(pika.compat.AbstractBase): @abc.abstractmethod - def abort(self): ... + def abort(self) -> None: ... @abc.abstractmethod - def get_protocol(self): ... + def get_protocol(self) -> AbstractStreamProtocol: ... @abc.abstractmethod - def write(self, data): ... + def write(self, data: bytes) -> None: ... @abc.abstractmethod - def get_write_buffer_size(self): ... + def get_write_buffer_size(self) -> int: ... diff --git a/stubs/pika/pika/adapters/utils/selector_ioloop_adapter.pyi b/stubs/pika/pika/adapters/utils/selector_ioloop_adapter.pyi index 17f1305f95fc..ae987ad316d1 100644 --- a/stubs/pika/pika/adapters/utils/selector_ioloop_adapter.pyi +++ b/stubs/pika/pika/adapters/utils/selector_ioloop_adapter.pyi @@ -1,78 +1,111 @@ import abc -from _typeshed import Incomplete +from collections.abc import Callable from logging import Logger +from typing import Final, Generic, Protocol, TypeVar, type_check_only from pika.adapters.utils import io_services_utils, nbio_interface LOGGER: Logger -class AbstractSelectorIOLoop(metaclass=abc.ABCMeta): +_Timeout = TypeVar("_Timeout", bound=object, default=object) + +@type_check_only +class _SupportsCancel(Protocol): + def cancel(self): ... + +class AbstractSelectorIOLoop(Generic[_Timeout], metaclass=abc.ABCMeta): @property @abc.abstractmethod - def READ(self): ... + def READ(self) -> int: ... @property @abc.abstractmethod - def WRITE(self): ... + def WRITE(self) -> int: ... @property @abc.abstractmethod - def ERROR(self): ... + def ERROR(self) -> int: ... @abc.abstractmethod - def close(self): ... + def close(self) -> None: ... @abc.abstractmethod - def start(self): ... + def start(self) -> None: ... @abc.abstractmethod - def stop(self): ... + def stop(self) -> None: ... @abc.abstractmethod - def call_later(self, delay, callback): ... + def call_later(self, delay: float, callback: Callable[[], object]) -> _Timeout: ... @abc.abstractmethod - def remove_timeout(self, timeout_handle): ... + def remove_timeout(self, timeout_handle: _Timeout) -> None: ... @abc.abstractmethod - def add_callback(self, callback): ... + def add_callback(self, callback: Callable[[], object]) -> None: ... @abc.abstractmethod - def add_handler(self, fd, handler, events): ... + def add_handler(self, fd: int, handler: Callable[[int, int], None], events: int) -> None: ... @abc.abstractmethod - def update_handler(self, fd, events): ... + def update_handler(self, fd: int, events: int) -> None: ... @abc.abstractmethod - def remove_handler(self, fd): ... + def remove_handler(self, fd: int) -> None: ... class SelectorIOServicesAdapter( io_services_utils.SocketConnectionMixin, io_services_utils.StreamingConnectionMixin, nbio_interface.AbstractIOServices, nbio_interface.AbstractFileDescriptorServices, + Generic[_Timeout], ): - def __init__(self, native_loop) -> None: ... - def get_native_ioloop(self): ... + def __init__(self, native_loop: AbstractSelectorIOLoop[_Timeout]) -> None: ... + def get_native_ioloop(self) -> AbstractSelectorIOLoop[_Timeout]: ... def close(self) -> None: ... def run(self) -> None: ... def stop(self) -> None: ... - def add_callback_threadsafe(self, callback) -> None: ... - def call_later(self, delay, callback): ... - def getaddrinfo(self, host, port, on_done, family: int = 0, socktype: int = 0, proto: int = 0, flags: int = 0): ... - def set_reader(self, fd, on_readable) -> None: ... - def remove_reader(self, fd): ... - def set_writer(self, fd, on_writable) -> None: ... - def remove_writer(self, fd): ... + def add_callback_threadsafe(self, callback: Callable[[], None]) -> None: ... + def call_later(self, delay: float, callback: Callable[[], None]) -> _TimerHandle: ... + def getaddrinfo( + self, + host: str | bytes | None, + port: str | bytes | int | None, + on_done: Callable[ # list is result of socket.getaddrinfo + [list[tuple[int, int, int, str, tuple[str, int] | tuple[str, int, int, int] | tuple[int, bytes]]] | BaseException], + None, + ], + family: int = 0, + socktype: int = 0, + proto: int = 0, + flags: int = 0, + ) -> nbio_interface.AbstractIOReference: ... + def set_reader(self, fd: int, on_readable: Callable[[], None]) -> None: ... + def remove_reader(self, fd: int) -> bool: ... + def set_writer(self, fd: int, on_writable: Callable[[], None]) -> None: ... + def remove_writer(self, fd: int) -> bool: ... class _FileDescriptorCallbacks: __slots__ = ("reader", "writer") - reader: Incomplete - writer: Incomplete - def __init__(self, reader=None, writer=None) -> None: ... + reader: Callable[[], None] + writer: Callable[[], None] + def __init__(self, reader: Callable[[], None] | None = None, writer: Callable[[], None] | None = None) -> None: ... class _TimerHandle(nbio_interface.AbstractTimerReference): - def __init__(self, handle, loop) -> None: ... + def __init__(self, handle: object, loop: AbstractSelectorIOLoop) -> None: ... def cancel(self) -> None: ... class _SelectorIOLoopIOHandle(nbio_interface.AbstractIOReference): - def __init__(self, subject) -> None: ... - def cancel(self): ... + def __init__(self, subject: _SupportsCancel) -> None: ... + def cancel(self) -> bool: ... class _AddressResolver: - NOT_STARTED: int - ACTIVE: int - CANCELED: int - COMPLETED: int - def __init__(self, native_loop, host, port, family, socktype, proto, flags, on_done) -> None: ... - def start(self): ... - def cancel(self): ... + NOT_STARTED: Final = 0 + ACTIVE: Final = 1 + CANCELED: Final = 2 + COMPLETED: Final = 3 + def __init__( + self, + native_loop: AbstractSelectorIOLoop, + host: str | bytes | None, + port: str | bytes | int | None, + family: int, + socktype: int, + proto: int, + flags: int, + on_done: Callable[ # list is result of socket.getaddrinfo + [list[tuple[int, int, int, str, tuple[str, int] | tuple[str, int, int, int] | tuple[int, bytes]]] | BaseException], + None, + ], + ) -> None: ... + def start(self) -> _SelectorIOLoopIOHandle: ... + def cancel(self) -> bool: ... From 89a7893de88b1c934e96c4bcbe47d11c188ea3b2 Mon Sep 17 00:00:00 2001 From: donbarbos Date: Tue, 21 Jul 2026 20:10:01 +0400 Subject: [PATCH 2/2] fix incompatible overrides --- stubs/pika/pika/adapters/asyncio_connection.pyi | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/stubs/pika/pika/adapters/asyncio_connection.pyi b/stubs/pika/pika/adapters/asyncio_connection.pyi index 3665ddc1a4f4..6b6b91ade128 100644 --- a/stubs/pika/pika/adapters/asyncio_connection.pyi +++ b/stubs/pika/pika/adapters/asyncio_connection.pyi @@ -46,9 +46,9 @@ class _AsyncioIOServicesAdapter( def call_later(self, delay: float, callback: Callable[[], object]) -> _TimerHandle: ... def getaddrinfo( self, - host: str, - port: int, - on_done: Callable[..., object], + host: str | bytes | None, + port: str | bytes | int | None, + on_done: Callable[[BaseConnection | BaseException], object], # type: ignore[override] family: int = 0, socktype: int = 0, proto: int = 0,