[pika] Complete adapters.utils (#16061)

Author: donbarbos

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,

stubs/pika/pika/adapters/utils/connection_workflow.pyi

@@ -1,15 +1,18 @@
 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): ...
 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: ...

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: ...

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: ...