diff --git a/stubs/pika/@tests/stubtest_allowlist.txt b/stubs/pika/@tests/stubtest_allowlist.txt index 30b2bf8aeb3c..218eda692ae1 100644 --- a/stubs/pika/@tests/stubtest_allowlist.txt +++ b/stubs/pika/@tests/stubtest_allowlist.txt @@ -12,3 +12,29 @@ pika.connection.ConnectionParameters.T # Arguments have a sentinel default, which is not reflected in the stubs. pika.connection.ConnectionParameters.__init__ + +# These are defined as None, but on initialization are set as callable attributes. +pika.adapters.base_connection._StreamingProtocolShim.connection_made +pika.adapters.base_connection._StreamingProtocolShim.connection_lost +pika.adapters.base_connection._StreamingProtocolShim.eof_received +pika.adapters.base_connection._StreamingProtocolShim.data_received + +# The following methods are not defined directly on this class; +# they are resolved via __getattr__. +pika.adapters.base_connection._StreamingProtocolShim.add_on_close_callback +pika.adapters.base_connection._StreamingProtocolShim.add_on_connection_blocked_callback +pika.adapters.base_connection._StreamingProtocolShim.add_on_connection_unblocked_callback +pika.adapters.base_connection._StreamingProtocolShim.add_on_open_callback +pika.adapters.base_connection._StreamingProtocolShim.add_on_open_error_callback +pika.adapters.base_connection._StreamingProtocolShim.channel +pika.adapters.base_connection._StreamingProtocolShim.update_secret +pika.adapters.base_connection._StreamingProtocolShim.close +pika.adapters.base_connection._StreamingProtocolShim.is_closed +pika.adapters.base_connection._StreamingProtocolShim.is_closing +pika.adapters.base_connection._StreamingProtocolShim.is_open +pika.adapters.base_connection._StreamingProtocolShim.basic_nack +pika.adapters.base_connection._StreamingProtocolShim.consumer_cancel_notify +pika.adapters.base_connection._StreamingProtocolShim.exchange_exchange_bindings +pika.adapters.base_connection._StreamingProtocolShim.publisher_confirms +pika.adapters.base_connection._StreamingProtocolShim.create_connection +pika.adapters.base_connection._StreamingProtocolShim.ioloop diff --git a/stubs/pika/METADATA.toml b/stubs/pika/METADATA.toml index d1f0f47d68cd..87c644180e73 100644 --- a/stubs/pika/METADATA.toml +++ b/stubs/pika/METADATA.toml @@ -5,6 +5,7 @@ extra-description = """\ The `types-pika` package contains alternate, more complete type stubs, that \ are maintained outside of typeshed.\ """ +optional-dependencies = ["types-gevent"] [tool.stubtest] stubtest-dependencies = ["gevent", "tornado", "twisted"] diff --git a/stubs/pika/pika/adapters/asyncio_connection.pyi b/stubs/pika/pika/adapters/asyncio_connection.pyi index 6b6b91ade128..7789841db6a9 100644 --- a/stubs/pika/pika/adapters/asyncio_connection.pyi +++ b/stubs/pika/pika/adapters/asyncio_connection.pyi @@ -1,25 +1,29 @@ +import asyncio from _typeshed import Incomplete -from asyncio import AbstractEventLoop, Future, Handle from collections.abc import Callable, Sequence from logging import Logger -from typing_extensions import Self -from ..connection import Connection, Parameters -from .base_connection import BaseConnection -from .utils import io_services_utils -from .utils.connection_workflow import AbstractAMQPConnectionWorkflow, AMQPConnectorException -from .utils.nbio_interface import AbstractFileDescriptorServices, AbstractIOReference, AbstractIOServices, AbstractTimerReference +from pika.adapters.base_connection import BaseConnection +from pika.adapters.utils import io_services_utils +from pika.adapters.utils.connection_workflow import AbstractAMQPConnectionWorkflow, AMQPConnectorException +from pika.adapters.utils.nbio_interface import ( + AbstractFileDescriptorServices, + AbstractIOReference, + AbstractIOServices, + AbstractTimerReference, +) +from pika.connection import Connection, Parameters LOGGER: Logger -class AsyncioConnection(BaseConnection): +class AsyncioConnection(BaseConnection[asyncio.AbstractEventLoop]): def __init__( self, parameters: Parameters | None = None, - on_open_callback: Callable[[Self], object] | None = None, - on_open_error_callback: Callable[[Self, BaseException], object] | None = None, - on_close_callback: Callable[[Self, BaseException], object] | None = None, - custom_ioloop: AbstractEventLoop | None = None, + on_open_callback: Callable[[Connection], object] | None = None, + on_open_error_callback: Callable[[Connection, BaseException], object] | None = None, + on_close_callback: Callable[[Connection, BaseException], object] | None = None, + custom_ioloop: asyncio.AbstractEventLoop | AbstractIOServices | None = None, internal_connection_workflow: bool = True, ) -> None: ... @classmethod @@ -27,7 +31,7 @@ class AsyncioConnection(BaseConnection): cls, connection_configs: Sequence[Parameters], on_done: Callable[[Connection | AMQPConnectorException], object], - custom_ioloop: AbstractEventLoop | None = None, + custom_ioloop: asyncio.AbstractEventLoop | None = None, workflow: AbstractAMQPConnectionWorkflow | None = None, ) -> AbstractAMQPConnectionWorkflow: ... @@ -37,8 +41,8 @@ class _AsyncioIOServicesAdapter( AbstractIOServices, AbstractFileDescriptorServices, ): - def __init__(self, loop: AbstractEventLoop | None = None) -> None: ... - def get_native_ioloop(self) -> AbstractEventLoop: ... + def __init__(self, loop: asyncio.AbstractEventLoop | None = None) -> None: ... + def get_native_ioloop(self) -> asyncio.AbstractEventLoop: ... def close(self) -> None: ... def run(self) -> None: ... def stop(self) -> None: ... @@ -48,7 +52,7 @@ class _AsyncioIOServicesAdapter( self, host: str | bytes | None, port: str | bytes | int | None, - on_done: Callable[[BaseConnection | BaseException], object], # type: ignore[override] + on_done: Callable[[BaseConnection[asyncio.AbstractEventLoop] | BaseException], object], # type: ignore[override] family: int = 0, socktype: int = 0, proto: int = 0, @@ -60,9 +64,13 @@ class _AsyncioIOServicesAdapter( def remove_writer(self, fd: int) -> bool: ... class _TimerHandle(AbstractTimerReference): - def __init__(self, handle: Handle) -> None: ... + def __init__(self, handle: asyncio.Handle) -> None: ... def cancel(self) -> None: ... class _AsyncioIOReference(AbstractIOReference): - def __init__(self, future: Future[Incomplete], on_done: Callable[[BaseConnection | BaseException], object]) -> None: ... + def __init__( + self, + future: asyncio.Future[Incomplete], + on_done: Callable[[BaseConnection[asyncio.AbstractEventLoop] | BaseException], object], + ) -> None: ... def cancel(self) -> bool: ... diff --git a/stubs/pika/pika/adapters/base_connection.pyi b/stubs/pika/pika/adapters/base_connection.pyi index 71468c334ed2..3e7e4866b60f 100644 --- a/stubs/pika/pika/adapters/base_connection.pyi +++ b/stubs/pika/pika/adapters/base_connection.pyi @@ -1,35 +1,112 @@ import abc from _typeshed import Incomplete -from collections.abc import Callable +from collections.abc import Callable, Mapping, Sequence from logging import Logger +from typing import Final, Generic, Literal, TypeVar from typing_extensions import Self -from ..adapters.utils.nbio_interface import AbstractIOServices, AbstractStreamProtocol -from ..connection import Connection, Parameters +from pika.adapters.utils.connection_workflow import AbstractAMQPConnectionWorkflow, AMQPConnectorException +from pika.adapters.utils.nbio_interface import AbstractIOServices, AbstractStreamProtocol, AbstractStreamTransport +from pika.callback import CallbackManager +from pika.channel import Channel +from pika.connection import Connection, Parameters +from pika.frame import Method +from pika.spec import Connection as SpecConnection LOGGER: Logger -class BaseConnection(Connection, metaclass=abc.ABCMeta): +_IOLoop = TypeVar("_IOLoop") + +class BaseConnection(Connection, Generic[_IOLoop], metaclass=abc.ABCMeta): def __init__( self, parameters: Parameters | None, - on_open_callback: Callable[[Self], object] | None, - on_open_error_callback: Callable[[Self, BaseException], object] | None, - on_close_callback: Callable[[Self, BaseException], object] | None, + on_open_callback: Callable[[Connection], object] | None, + on_open_error_callback: Callable[[Connection, BaseException], object] | None, + on_close_callback: Callable[[Connection, BaseException], object] | None, nbio: AbstractIOServices, internal_connection_workflow: bool = True, ) -> None: ... @classmethod @abc.abstractmethod - def create_connection(cls, connection_configs, on_done, custom_ioloop=None, workflow=None): ... + def create_connection( + cls, + connection_configs: Sequence[Parameters], + on_done: Callable[[Connection | AMQPConnectorException], object], + custom_ioloop: _IOLoop | None = None, + workflow: AbstractAMQPConnectionWorkflow | None = None, + ) -> AbstractAMQPConnectionWorkflow: ... + @property + def ioloop(self) -> _IOLoop: ... + +class _StreamingProtocolShim(AbstractStreamProtocol, Generic[_IOLoop]): + conn: BaseConnection[_IOLoop] + def __init__(self, conn: BaseConnection[_IOLoop]) -> None: ... + # These are defined as None, but on initialization are set as callable attributes + def connection_made(self, transport: AbstractStreamTransport) -> None: ... + def connection_lost(self, error: BaseException | None) -> None: ... + def eof_received(self) -> bool: ... + def data_received(self, data: bytes) -> None: ... + + # Next attributes are accessed via getattr() from connection.Connection class: + ON_CONNECTION_CLOSED: Final = "_on_connection_closed" + ON_CONNECTION_ERROR: Final = "_on_connection_error" + ON_CONNECTION_OPEN_OK: Final = "_on_connection_open_ok" + CONNECTION_CLOSED: Final = 0 + CONNECTION_INIT: Final = 1 + CONNECTION_PROTOCOL: Final = 2 + CONNECTION_START: Final = 3 + CONNECTION_TUNE: Final = 4 + CONNECTION_OPEN: Final = 5 + CONNECTION_CLOSING: Final = 6 + connection_state: Literal[0, 1, 2, 3, 4, 5, 6] # one of the constants above + params: Parameters + callbacks: CallbackManager + server_capabilities: Mapping[str, bool] | None + server_properties: Mapping[str, Incomplete] | None + known_hosts: str | None + def add_on_close_callback(self, callback: Callable[[Self, BaseException], object]) -> None: ... + def add_on_connection_blocked_callback(self, callback: Callable[[Self, Method[SpecConnection.Blocked]], object]) -> None: ... + def add_on_connection_unblocked_callback( + self, callback: Callable[[Self, Method[SpecConnection.Unblocked]], object] + ) -> None: ... + def add_on_open_callback(self, callback: Callable[[Self], object]) -> None: ... + def add_on_open_error_callback( + self, callback: Callable[[Self, BaseException], object], remove_default: bool = True + ) -> None: ... + def channel( + self, channel_number: int | None = None, on_open_callback: Callable[[Channel], object] | None = None + ) -> Channel: ... + def update_secret( + self, + new_secret: str | bytes, + reason: str | bytes, + callback: Callable[[Method[SpecConnection.UpdateSecretOk]], object] | None = None, + ) -> None: ... + def close(self, reply_code: int = 200, reply_text: str = "Normal shutdown") -> None: ... @property - def ioloop(self): ... + def is_closed(self) -> bool: ... + @property + def is_closing(self) -> bool: ... + @property + def is_open(self) -> bool: ... + @property + def basic_nack(self) -> bool: ... + @property + def consumer_cancel_notify(self) -> bool: ... + @property + def exchange_exchange_bindings(self) -> bool: ... + @property + def publisher_confirms(self) -> bool: ... -class _StreamingProtocolShim(AbstractStreamProtocol): - connection_made: Incomplete - connection_lost: Incomplete - eof_received: Incomplete - data_received: Incomplete - conn: Incomplete - def __init__(self, conn) -> None: ... - def __getattr__(self, attr: str): ... + # Next attributes are accessed via getattr() from BaseConnection class: + @classmethod + def create_connection( + cls, + connection_configs: Sequence[Parameters], + on_done: Callable[[Connection | AMQPConnectorException], object], + custom_ioloop: _IOLoop | None = None, + workflow: AbstractAMQPConnectionWorkflow | None = None, + ) -> AbstractAMQPConnectionWorkflow: ... + @property + def ioloop(self) -> _IOLoop: ... diff --git a/stubs/pika/pika/adapters/blocking_connection.pyi b/stubs/pika/pika/adapters/blocking_connection.pyi index 9aabeb7befa8..5dd25332bacb 100644 --- a/stubs/pika/pika/adapters/blocking_connection.pyi +++ b/stubs/pika/pika/adapters/blocking_connection.pyi @@ -1,79 +1,66 @@ -from _typeshed import Incomplete, Unused -from collections.abc import Callable, Generator, Sequence +from _typeshed import Incomplete +from collections import deque +from collections.abc import Callable, Generator, Mapping, Sequence from logging import Logger from types import TracebackType -from typing import Any, NamedTuple, TypeVar +from typing import Final, Generic, TypeVar, overload from typing_extensions import Self -from ..connection import Parameters -from ..exchange_type import ExchangeType -from ..frame import Method -from ..spec import Basic, BasicProperties, Connection, Exchange, Queue, Tx +from pika import connection +from pika.adapters.select_connection import SelectConnection +from pika.channel import Channel +from pika.exchange_type import ExchangeType +from pika.frame import Method +from pika.spec import Basic, BasicProperties, Connection, Exchange, Queue, Tx T = TypeVar("T", bound=Connection.Blocked | Connection.Unblocked) # noqa: Y001 LOGGER: Logger -class _CallbackResult: - __slots__ = ("_value_class", "_ready", "_values") - def __init__(self, value_class=None) -> None: ... - def reset(self) -> None: ... - def __bool__(self) -> bool: ... - __nonzero__: Incomplete - def __enter__(self): ... - def __exit__(self, *args: Unused, **kwargs: Unused) -> None: ... - def is_ready(self): ... - @property - def ready(self): ... - def signal_once(self, *_args, **_kwargs) -> None: ... - def set_value_once(self, *args, **kwargs) -> None: ... - def append_element(self, *args, **kwargs) -> None: ... - @property - def value(self): ... - @property - def elements(self): ... - class _IoloopTimerContext: - def __init__(self, duration, connection) -> None: ... - def __enter__(self): ... - def __exit__(self, *_args: Unused, **_kwargs: Unused) -> None: ... - def is_ready(self): ... + def __init__(self, duration: float, connection: SelectConnection) -> None: ... + def __enter__(self) -> Self: ... + def __exit__( + self, exc_type: type[BaseException] | None, value: BaseException | None, traceback: TracebackType | None + ) -> None: ... + def is_ready(self) -> bool: ... class _TimerEvt: __slots__ = ("timer_id", "_callback") - timer_id: Incomplete - def __init__(self, callback) -> None: ... + timer_id: object | None + def __init__(self, callback: Callable[[], None]) -> None: ... def dispatch(self) -> None: ... -class _ConnectionBlockedUnblockedEvtBase: +class _ConnectionBlockedUnblockedEvtBase(Generic[T]): __slots__ = ("_callback", "_method_frame") - def __init__(self, callback, method_frame) -> None: ... + def __init__(self, callback: Callable[[Method[T]], None], method_frame: Method[T]) -> None: ... def dispatch(self) -> None: ... -class _ConnectionBlockedEvt(_ConnectionBlockedUnblockedEvtBase): ... -class _ConnectionUnblockedEvt(_ConnectionBlockedUnblockedEvtBase): ... +class _ConnectionBlockedEvt(_ConnectionBlockedUnblockedEvtBase[Connection.Blocked]): ... +class _ConnectionUnblockedEvt(_ConnectionBlockedUnblockedEvtBase[Connection.Unblocked]): ... class BlockingConnection: - class _OnClosedArgs(NamedTuple): - connection: Incomplete - error: Incomplete - - class _OnChannelOpenedArgs(NamedTuple): - channel: Incomplete - - def __init__(self, parameters: Parameters | Sequence[Parameters] | None = None, _impl_class=None) -> None: ... + def __init__( + self, + parameters: connection.Parameters | Sequence[connection.Parameters] | None = None, + _impl_class: SelectConnection | None = None, + ) -> None: ... def __enter__(self) -> Self: ... def __exit__( self, exc_type: type[BaseException] | None, value: BaseException | None, traceback: TracebackType | None ) -> None: ... - def add_on_connection_blocked_callback(self, callback) -> None: ... - def add_on_connection_unblocked_callback(self, callback) -> None: ... - def call_later(self, delay, callback): ... - def add_callback_threadsafe(self, callback) -> None: ... - def remove_timeout(self, timeout_id) -> None: ... - def update_secret(self, new_secret, reason) -> None: ... + def add_on_connection_blocked_callback( + self, callback: Callable[[connection.Connection, Method[Connection.Blocked]], None] + ) -> None: ... + def add_on_connection_unblocked_callback( + self, callback: Callable[[connection.Connection, Method[Connection.Unblocked]], None] + ) -> None: ... + def call_later(self, delay: float, callback: Callable[[], None]) -> object: ... + def add_callback_threadsafe(self, callback: Callable[..., None]) -> None: ... + def remove_timeout(self, timeout_id: object) -> None: ... + def update_secret(self, new_secret: str, reason: str) -> None: ... def close(self, reply_code: int = 200, reply_text: str = "Normal shutdown") -> None: ... - def process_data_events(self, time_limit: int | None = 0) -> None: ... + def process_data_events(self, time_limit: float | None = 0) -> None: ... def sleep(self, duration: float) -> None: ... def channel(self, channel_number: int | None = None) -> BlockingChannel: ... @property @@ -97,127 +84,153 @@ class _ChannelPendingEvt: ... class _ConsumerDeliveryEvt(_ChannelPendingEvt): __slots__ = ("method", "properties", "body") - method: Incomplete - properties: Incomplete - body: Incomplete - def __init__(self, method, properties, body) -> None: ... + method: Basic.Deliver + properties: BasicProperties + body: bytes + def __init__(self, method: Basic.Deliver, properties: BasicProperties, body: bytes) -> None: ... class _ConsumerCancellationEvt(_ChannelPendingEvt): __slots__ = ("method_frame",) - method_frame: Incomplete - def __init__(self, method_frame) -> None: ... + method_frame: Method[Basic.Cancel] + def __init__(self, method_frame: Method[Basic.Cancel]) -> None: ... @property - def method(self): ... + def method(self) -> Basic.Cancel: ... class _ReturnedMessageEvt(_ChannelPendingEvt): __slots__ = ("callback", "channel", "method", "properties", "body") - callback: Incomplete - channel: Incomplete - method: Incomplete - properties: Incomplete - body: Incomplete - def __init__(self, callback, channel, method, properties, body) -> None: ... + callback: Callable[[BlockingChannel, Basic.Return, BasicProperties, bytes], None] + channel: BlockingChannel + method: Basic.Return + properties: BasicProperties + body: bytes + def __init__( + self, + callback: Callable[[BlockingChannel, Basic.Return, BasicProperties, bytes], None], + channel: BlockingChannel, + method: Basic.Return, + properties: BasicProperties, + body: bytes, + ) -> None: ... def dispatch(self) -> None: ... class ReturnedMessage: __slots__ = ("method", "properties", "body") - method: Incomplete - properties: Incomplete - body: Incomplete - def __init__(self, method, properties, body) -> None: ... + method: Basic.Return + properties: BasicProperties + body: bytes + def __init__(self, method: Basic.Return, properties: BasicProperties, body: bytes) -> None: ... class _ConsumerInfo: __slots__ = ("consumer_tag", "auto_ack", "on_message_callback", "alternate_event_sink", "state") - SETTING_UP: int - ACTIVE: int - TEARING_DOWN: int - CANCELLED_BY_BROKER: int - consumer_tag: Incomplete - auto_ack: Incomplete - on_message_callback: Incomplete - alternate_event_sink: Incomplete - state: Incomplete - def __init__(self, consumer_tag, auto_ack, on_message_callback=None, alternate_event_sink=None) -> None: ... + SETTING_UP: Final = 1 + ACTIVE: Final = 2 + TEARING_DOWN: Final = 3 + CANCELLED_BY_BROKER: Final = 4 + consumer_tag: str + auto_ack: bool + on_message_callback: Callable[[BlockingChannel, Basic.Deliver, BasicProperties, bytes], None] | None + alternate_event_sink: Callable[[_ChannelPendingEvt], None] | None + state: int + + @overload + def __init__( + self, + consumer_tag: str, + auto_ack: bool, + # Only one of them must be non-None: + on_message_callback: Callable[[BlockingChannel, Basic.Deliver, BasicProperties, bytes], None], + alternate_event_sink: None = None, + ) -> None: ... + @overload + def __init__( + self, + consumer_tag: str, + auto_ack: bool, + # Only one of them must be non-None: + on_message_callback: None = None, + alternate_event_sink: Callable[[_ChannelPendingEvt], None] = ..., + ) -> None: ... + @property - def setting_up(self): ... + def setting_up(self) -> bool: ... @property - def active(self): ... + def active(self) -> bool: ... @property - def tearing_down(self): ... + def tearing_down(self) -> bool: ... @property - def cancelled_by_broker(self): ... + def cancelled_by_broker(self) -> bool: ... class _QueueConsumerGeneratorInfo: __slots__ = ("params", "consumer_tag", "pending_events") - params: Incomplete - consumer_tag: Incomplete - pending_events: Incomplete - def __init__(self, params, consumer_tag) -> None: ... + params: tuple[str, bool, bool] + consumer_tag: str + pending_events: deque[_ChannelPendingEvt] + def __init__(self, params: tuple[str, bool, bool], consumer_tag: str) -> None: ... class BlockingChannel: - class _RxMessageArgs(NamedTuple): - channel: Incomplete - method: Incomplete - properties: Incomplete - body: Incomplete - - class _MethodFrameCallbackResultArgs(NamedTuple): - method_frame: Incomplete - - class _OnMessageConfirmationReportArgs(NamedTuple): - method_frame: Incomplete - - class _FlowOkCallbackResultArgs(NamedTuple): - active: Incomplete - - def __init__(self, channel_impl, connection) -> None: ... + def __init__(self, channel_impl: Channel, connection: BlockingConnection) -> None: ... def __int__(self) -> int: ... - def __enter__(self): ... + def __enter__(self) -> Self: ... def __exit__( self, exc_type: type[BaseException] | None, value: BaseException | None, traceback: TracebackType | None ) -> None: ... @property - def channel_number(self): ... + def channel_number(self) -> int: ... @property - def connection(self): ... + def connection(self) -> BlockingConnection: ... @property - def is_closed(self): ... + def is_closed(self) -> bool: ... @property - def is_open(self): ... + def is_open(self) -> bool: ... @property - def consumer_tags(self): ... - def close(self, reply_code: int = 0, reply_text: str = "Normal shutdown"): ... - def flow(self, active): ... - def add_on_cancel_callback(self, callback) -> None: ... - def add_on_return_callback(self, callback): ... + def consumer_tags(self) -> list[str]: ... + def close(self, reply_code: int = 0, reply_text: str = "Normal shutdown") -> None: ... + def flow(self, active: bool) -> bool: ... + def add_on_cancel_callback(self, callback: Callable[[Method[Basic.Cancel]], None]) -> None: ... + def add_on_return_callback( + self, callback: Callable[[BlockingChannel, Basic.Return, BasicProperties, bytes], None] + ) -> None: ... def basic_consume( self, queue: str, - on_message_callback: Callable[[BlockingChannel, Basic.Deliver, BasicProperties, bytes], object], + on_message_callback: Callable[[BlockingChannel, Basic.Deliver, BasicProperties, bytes], None], auto_ack: bool = False, exclusive: bool = False, consumer_tag: str | None = None, - arguments: dict[str, Any] | None = None, + arguments: Mapping[str, Incomplete] | None = None, ) -> str: ... - def basic_cancel(self, consumer_tag: str) -> Sequence[tuple[Basic.Deliver, BasicProperties, bytes]]: ... + def basic_cancel(self, consumer_tag: str) -> list[tuple[Basic.Deliver, BasicProperties, bytes]]: ... def start_consuming(self) -> None: ... def stop_consuming(self, consumer_tag: str | None = None) -> None: ... + + @overload def consume( self, queue: str, auto_ack: bool = False, exclusive: bool = False, - arguments: dict[str, Any] | None = None, - inactivity_timeout: float | None = None, + arguments: Mapping[str, Incomplete] | None = None, + inactivity_timeout: None = None, # different between overloads consumer_tag: str | None = None, - ) -> Generator[tuple[Basic.Deliver | None, BasicProperties | None, bytes | None]]: ... + ) -> Generator[tuple[Basic.Deliver, BasicProperties, bytes]]: ... + @overload + def consume( + self, + queue: str, + auto_ack: bool = False, + exclusive: bool = False, + arguments: Mapping[str, Incomplete] | None = None, + inactivity_timeout: float = ..., # different between overloads + consumer_tag: str | None = None, + ) -> Generator[tuple[Basic.Deliver, BasicProperties, bytes] | tuple[None, None, None]]: ... + def get_waiting_message_count(self) -> int: ... def cancel(self) -> int: ... def basic_ack(self, delivery_tag: int = 0, multiple: bool = False) -> None: ... def basic_nack(self, delivery_tag: int = 0, multiple: bool = False, requeue: bool = True) -> None: ... def basic_get( self, queue: str, auto_ack: bool = False - ) -> tuple[Basic.GetOk | None, BasicProperties | None, bytes | None]: ... + ) -> tuple[Basic.GetOk, BasicProperties, bytes] | tuple[None, None, None]: ... def basic_publish( self, exchange: str, @@ -238,18 +251,18 @@ class BlockingChannel: durable: bool = False, auto_delete: bool = False, internal: bool = False, - arguments: dict[str, Any] | None = None, + arguments: Mapping[str, Incomplete] | None = None, ) -> None: ... def exchange_delete(self, exchange: str | None = None, if_unused: bool = False) -> Method[Exchange.DeleteOk]: ... def exchange_bind( - self, destination: str, source: str, routing_key: str = "", arguments: dict[str, Any] | None = None + self, destination: str, source: str, routing_key: str = "", arguments: Mapping[str, Incomplete] | None = None ) -> Method[Exchange.BindOk]: ... def exchange_unbind( self, destination: str | None = None, source: str | None = None, routing_key: str = "", - arguments: dict[str, Any] | None = None, + arguments: Mapping[str, Incomplete] | None = None, ) -> Method[Exchange.UnbindOk]: ... def queue_declare( self, @@ -258,15 +271,15 @@ class BlockingChannel: durable: bool = False, exclusive: bool = False, auto_delete: bool = False, - arguments: dict[str, Any] | None = None, + arguments: Mapping[str, Incomplete] | None = None, ) -> Method[Queue.DeclareOk]: ... def queue_delete(self, queue: str, if_unused: bool = False, if_empty: bool = False) -> Method[Queue.DeleteOk]: ... def queue_purge(self, queue: str) -> Method[Queue.PurgeOk]: ... def queue_bind( - self, queue: str, exchange: str, routing_key: str | None = None, arguments: dict[str, Any] | None = None + self, queue: str, exchange: str, routing_key: str | None = None, arguments: Mapping[str, Incomplete] | None = None ) -> Method[Queue.BindOk]: ... def queue_unbind( - self, queue: str, exchange: str, routing_key: str | None = None, arguments: dict[str, Any] | None = None + self, queue: str, exchange: str, routing_key: str | None = None, arguments: Mapping[str, Incomplete] | None = None ) -> Method[Queue.UnbindOk]: ... def tx_select(self) -> Method[Tx.SelectOk]: ... def tx_commit(self) -> Method[Tx.CommitOk]: ... diff --git a/stubs/pika/pika/adapters/gevent_connection.pyi b/stubs/pika/pika/adapters/gevent_connection.pyi index ab30dd6c4f83..1f323796d6c1 100644 --- a/stubs/pika/pika/adapters/gevent_connection.pyi +++ b/stubs/pika/pika/adapters/gevent_connection.pyi @@ -1,55 +1,92 @@ +from collections.abc import Callable, Sequence from logging import Logger +from typing import Final +from gevent._types import _Loop, _TimerWatcher +from gevent.hub import Hub from pika.adapters.base_connection import BaseConnection -from pika.adapters.utils.nbio_interface import AbstractIOReference -from pika.adapters.utils.selector_ioloop_adapter import AbstractSelectorIOLoop, SelectorIOServicesAdapter +from pika.adapters.utils.connection_workflow import AbstractAMQPConnectionWorkflow, AMQPConnectorException +from pika.adapters.utils.nbio_interface import AbstractIOReference, AbstractIOServices +from pika.adapters.utils.selector_ioloop_adapter import AbstractSelectorIOLoop, SelectorIOServicesAdapter, _SupportsCancel +from pika.connection import Connection, Parameters LOGGER: Logger -class GeventConnection(BaseConnection): +class GeventConnection(BaseConnection[_Loop]): def __init__( self, - parameters=None, - on_open_callback=None, - on_open_error_callback=None, - on_close_callback=None, - custom_ioloop=None, + parameters: Parameters | None = None, + on_open_callback: Callable[[Connection], object] | None = None, + on_open_error_callback: Callable[[Connection, BaseException], object] | None = None, + on_close_callback: Callable[[Connection, BaseException], object] | None = None, + custom_ioloop: _Loop | AbstractIOServices | None = None, internal_connection_workflow: bool = True, ) -> None: ... @classmethod - def create_connection(cls, connection_configs, on_done, custom_ioloop=None, workflow=None): ... + def create_connection( + cls, + connection_configs: Sequence[Parameters], + on_done: Callable[[Connection | AMQPConnectorException], object], + custom_ioloop: _Loop | None = None, + workflow: AbstractAMQPConnectionWorkflow | None = None, + ) -> AbstractAMQPConnectionWorkflow: ... class _TSafeCallbackQueue: def __init__(self) -> None: ... @property - def fd(self): ... - def add_callback_threadsafe(self, callback) -> None: ... + def fd(self) -> int: ... + def add_callback_threadsafe(self, callback: Callable[[], None]) -> None: ... def run_next_callback(self) -> None: ... -class _GeventSelectorIOLoop(AbstractSelectorIOLoop): - READ: int - WRITE: int - ERROR: int - def __init__(self, gevent_hub=None) -> None: ... +class _GeventSelectorIOLoop(AbstractSelectorIOLoop[_TimerWatcher]): + READ: Final[int] + WRITE: Final[int] + ERROR: Final[int] + def __init__(self, gevent_hub: Hub | None = None) -> None: ... def close(self) -> None: ... def start(self) -> None: ... def stop(self) -> None: ... - def add_callback(self, callback) -> None: ... - def call_later(self, delay, callback): ... - def remove_timeout(self, timeout_handle) -> None: ... - def add_handler(self, fd, handler, events) -> None: ... - def update_handler(self, fd, events) -> None: ... - def remove_handler(self, fd) -> None: ... + def add_callback(self, callback: Callable[[], object]) -> None: ... + def call_later(self, delay: float, callback: Callable[[], object]) -> _TimerWatcher: ... + def remove_timeout(self, timeout_handle: _TimerWatcher) -> None: ... + def add_handler(self, fd: int, handler: Callable[[int, int], None], events: int) -> None: ... + def update_handler(self, fd: int, events: int) -> None: ... + def remove_handler(self, fd: int) -> None: ... class _GeventSelectorIOServicesAdapter(SelectorIOServicesAdapter): - 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: ... class _GeventIOLoopIOHandle(AbstractIOReference): - def __init__(self, subject) -> None: ... - def cancel(self): ... + def __init__(self, subject: _SupportsCancel) -> None: ... + def cancel(self) -> bool: ... class _GeventAddressResolver: __slots__ = ("_loop", "_on_done", "_greenlet", "_ga_host", "_ga_port", "_ga_family", "_ga_socktype", "_ga_proto", "_ga_flags") - def __init__(self, native_loop, host, port, family, socktype, proto, flags, on_done) -> None: ... + 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) -> None: ... - def cancel(self): ... + def cancel(self) -> bool: ... diff --git a/stubs/pika/pika/adapters/select_connection.pyi b/stubs/pika/pika/adapters/select_connection.pyi index 86ef12c83e88..d1309477191c 100644 --- a/stubs/pika/pika/adapters/select_connection.pyi +++ b/stubs/pika/pika/adapters/select_connection.pyi @@ -1,54 +1,62 @@ import abc import select -from _typeshed import Incomplete -from collections.abc import Callable +from collections.abc import Callable, Sequence from logging import Logger -from typing import Final, Literal, TypeAlias, TypedDict +from typing import ClassVar, Final, Literal, TypeAlias, TypedDict import pika.compat from pika.adapters.base_connection import BaseConnection +from pika.adapters.utils.connection_workflow import AbstractAMQPConnectionWorkflow, AMQPConnectorException +from pika.adapters.utils.nbio_interface import AbstractIOServices from pika.adapters.utils.selector_ioloop_adapter import AbstractSelectorIOLoop +from pika.connection import Connection, Parameters -SELECT_ERROR_T: TypeAlias = OSError | InterruptedError | select.error +SELECT_ERROR_T: TypeAlias = OSError | IOError | InterruptedError | select.error class POLLER_PARAMS(TypedDict): get_wait_seconds: Callable[[], float | None] - process_timeouts: Callable[[], object] + process_timeouts: Callable[[], None] LOGGER: Logger SELECT_TYPE: Literal["epoll", "kqueue", "poll"] | None -class SelectConnection(BaseConnection): +class SelectConnection(BaseConnection[IOLoop]): def __init__( self, - parameters=None, - on_open_callback=None, - on_open_error_callback=None, - on_close_callback=None, - custom_ioloop=None, + parameters: Parameters | None = None, + on_open_callback: Callable[[Connection], object] | None = None, + on_open_error_callback: Callable[[Connection, BaseException], object] | None = None, + on_close_callback: Callable[[Connection, BaseException], object] | None = None, + custom_ioloop: IOLoop | AbstractIOServices | None = None, internal_connection_workflow: bool = True, ) -> None: ... @classmethod - def create_connection(cls, connection_configs, on_done, custom_ioloop=None, workflow=None): ... + def create_connection( + cls, + connection_configs: Sequence[Parameters], + on_done: Callable[[Connection | AMQPConnectorException], object], + custom_ioloop: IOLoop | None = None, + workflow: AbstractAMQPConnectionWorkflow | None = None, + ) -> AbstractAMQPConnectionWorkflow: ... class _Timeout: __slots__ = ("deadline", "callback") - deadline: Incomplete - callback: Incomplete - def __init__(self, deadline, callback) -> None: ... - def __eq__(self, other): ... - def __ne__(self, other): ... - def __lt__(self, other): ... - def __gt__(self, other): ... - def __le__(self, other): ... - def __ge__(self, other): ... + deadline: float + callback: Callable[[], None] + def __init__(self, deadline: float, callback: Callable[[], None]) -> None: ... + def __eq__(self, other: object) -> bool: ... + def __ne__(self, other: object) -> bool: ... + def __lt__(self, other: object) -> bool: ... + def __gt__(self, other: object) -> bool: ... + def __le__(self, other: object) -> bool: ... + def __ge__(self, other: object) -> bool: ... class _Timer: def __init__(self) -> None: ... def close(self) -> None: ... - def call_later(self, delay, callback): ... - def remove_timeout(self, timeout) -> None: ... - def get_remaining_interval(self): ... + def call_later(self, delay: float, callback: Callable[[], None]) -> _Timeout: ... + def remove_timeout(self, timeout: _Timeout) -> None: ... + def get_remaining_interval(self) -> float | None: ... def process_timeouts(self) -> None: ... class PollEvents: @@ -57,20 +65,20 @@ class PollEvents: ERROR: Final[int] HANGUP: Final[int] -class IOLoop(AbstractSelectorIOLoop): - READ: Incomplete - WRITE: Incomplete - ERROR: Incomplete +class IOLoop(AbstractSelectorIOLoop[_Timeout]): + READ: Final[int] + WRITE: Final[int] + ERROR: Final[int] def __init__(self) -> None: ... def close(self) -> None: ... - def call_later(self, delay, callback): ... - def remove_timeout(self, timeout_handle) -> None: ... - def add_callback_threadsafe(self, callback) -> None: ... - add_callback: Incomplete + def call_later(self, delay: float, callback: Callable[[], object]) -> _Timeout: ... + def remove_timeout(self, timeout_handle: _Timeout) -> None: ... + def add_callback_threadsafe(self, callback: Callable[[], object]) -> None: ... + add_callback = add_callback_threadsafe def process_timeouts(self) -> None: ... - def add_handler(self, fd, handler, events) -> None: ... - def update_handler(self, fd, events) -> None: ... - def remove_handler(self, fd) -> None: ... + def add_handler(self, fd: int, handler: Callable[[int, int], None], events: int) -> None: ... + def update_handler(self, fd: int, events: int) -> None: ... + def remove_handler(self, fd: int) -> None: ... def start(self) -> None: ... def stop(self) -> None: ... def activate_poller(self) -> None: ... @@ -78,32 +86,32 @@ class IOLoop(AbstractSelectorIOLoop): def poll(self) -> None: ... class _PollerBase(pika.compat.AbstractBase, metaclass=abc.ABCMeta): - POLL_TIMEOUT_MULT: int - def __init__(self, get_wait_seconds, process_timeouts) -> None: ... + POLL_TIMEOUT_MULT: ClassVar[int] + def __init__(self, get_wait_seconds: Callable[[], float | None], process_timeouts: Callable[[], None]) -> None: ... def close(self) -> None: ... def wake_threadsafe(self) -> None: ... - def add_handler(self, fileno, handler, events) -> None: ... - def update_handler(self, fileno, events) -> None: ... - def remove_handler(self, fileno) -> None: ... + def add_handler(self, fileno: int, handler: Callable[[int, int], None], events: int) -> None: ... + def update_handler(self, fileno: int, events: int) -> None: ... + def remove_handler(self, fileno: int) -> None: ... def activate_poller(self) -> None: ... def deactivate_poller(self) -> None: ... def start(self) -> None: ... def stop(self) -> None: ... @abc.abstractmethod - def poll(self): ... + def poll(self) -> None: ... class SelectPoller(_PollerBase): - POLL_TIMEOUT_MULT: int + POLL_TIMEOUT_MULT: ClassVar[int] def poll(self) -> None: ... class KQueuePoller(_PollerBase): - def __init__(self, get_wait_seconds, process_timeouts) -> None: ... + def __init__(self, get_wait_seconds: Callable[[], float | None], process_timeouts: Callable[[], None]) -> None: ... def poll(self) -> None: ... class PollPoller(_PollerBase): - POLL_TIMEOUT_MULT: int - def __init__(self, get_wait_seconds, process_timeouts) -> None: ... + POLL_TIMEOUT_MULT: ClassVar[int] + def __init__(self, get_wait_seconds: Callable[[], float | None], process_timeouts: Callable[[], None]) -> None: ... def poll(self) -> None: ... class EPollPoller(PollPoller): - POLL_TIMEOUT_MULT: int + POLL_TIMEOUT_MULT: ClassVar[int] diff --git a/stubs/pika/pika/adapters/tornado_connection.pyi b/stubs/pika/pika/adapters/tornado_connection.pyi index 9207ef355d0e..31ac6c9518d9 100644 --- a/stubs/pika/pika/adapters/tornado_connection.pyi +++ b/stubs/pika/pika/adapters/tornado_connection.pyi @@ -1,21 +1,32 @@ from _typeshed import Incomplete +from collections.abc import Callable, Sequence from logging import Logger +from typing import TypeAlias -from pika.adapters import base_connection +from pika.adapters.base_connection import BaseConnection +from pika.adapters.utils.connection_workflow import AbstractAMQPConnectionWorkflow, AMQPConnectorException +from pika.adapters.utils.nbio_interface import AbstractIOServices +from pika.connection import Connection, Parameters + +_IOLoop: TypeAlias = Incomplete # actual type is tornado.ioloop.IOLoop LOGGER: Logger -class TornadoConnection(base_connection.BaseConnection): +class TornadoConnection(BaseConnection[_IOLoop]): def __init__( self, - parameters: Incomplete | None = ..., - on_open_callback: Incomplete | None = ..., - on_open_error_callback: Incomplete | None = ..., - on_close_callback: Incomplete | None = ..., - custom_ioloop: Incomplete | None = ..., - internal_connection_workflow: bool = ..., + parameters: Parameters | None = None, + on_open_callback: Callable[[Connection], object] | None = None, + on_open_error_callback: Callable[[Connection, BaseException], object] | None = None, + on_close_callback: Callable[[Connection, BaseException], object] | None = None, + custom_ioloop: _IOLoop | AbstractIOServices | None = None, + internal_connection_workflow: bool = True, ) -> None: ... @classmethod def create_connection( - cls, connection_configs, on_done, custom_ioloop: Incomplete | None = ..., workflow: Incomplete | None = ... - ): ... + cls, + connection_configs: Sequence[Parameters], + on_done: Callable[[Connection | AMQPConnectorException], object], + custom_ioloop: _IOLoop | None = None, + workflow: AbstractAMQPConnectionWorkflow | None = None, + ) -> AbstractAMQPConnectionWorkflow: ... diff --git a/stubs/pika/pika/adapters/twisted_connection.pyi b/stubs/pika/pika/adapters/twisted_connection.pyi index 9d8d90ef333e..8b629dc7016d 100644 --- a/stubs/pika/pika/adapters/twisted_connection.pyi +++ b/stubs/pika/pika/adapters/twisted_connection.pyi @@ -2,12 +2,20 @@ # We don't want to force it as a dependency but that means we also can't test it with type-checkers given the current setup. from _typeshed import Incomplete -from collections.abc import Callable +from collections.abc import Callable, Iterable, Mapping from logging import Logger -from typing import Any, Generic, NamedTuple, TypeVar +from typing import Generic, NamedTuple, TypeVar +from pika import amqp_object from pika.adapters.utils.nbio_interface import AbstractTimerReference -from twisted.internet.base import DelayedCall # type: ignore[import-not-found] # pyright: ignore[reportMissingImports] +from pika.channel import Channel +from pika.connection import Connection, ConnectionParameters, Parameters +from pika.exchange_type import ExchangeType +from pika.spec import BasicProperties +from twisted.internet.base import ( # type: ignore[import-not-found] # pyright: ignore[reportMissingImports] + DelayedCall, + ReactorBase, +) from twisted.internet.defer import ( # type: ignore[import-not-found] # pyright: ignore[reportMissingImports] Deferred, DeferredQueue, @@ -16,107 +24,110 @@ from twisted.internet.interfaces import ITransport # type: ignore[import-not-fo from twisted.internet.protocol import Protocol # type: ignore[import-not-found] # pyright: ignore[reportMissingImports] from twisted.python.failure import Failure # type: ignore[import-not-found] # pyright: ignore[reportMissingImports] -from ..connection import Connection, Parameters +LOGGER: Logger _T = TypeVar("_T") -LOGGER: Logger - class ClosableDeferredQueue(DeferredQueue[_T], Generic[_T]): # pyright: ignore[reportUntypedBaseClass] # noqa: Y060 closed: Failure | BaseException | None - def __init__(self, size: Incomplete | None = ..., backlog: Incomplete | None = ...) -> None: ... + def __init__(self, size: int | None = None, backlog: int | None = None) -> None: ... # Returns a Deferred with an error if fails. None if success def put(self, obj: _T) -> Deferred[Failure | BaseException] | None: ... # type: ignore[override] # ignore is not needed for mypy, but is for stubtest def get(self) -> Deferred[Failure | BaseException | _T]: ... # type: ignore[override] # ignore is not needed for mypy, but is for stubtest - pending: Incomplete - def close(self, reason: BaseException | None) -> None: ... + pending: list[_T] + def close(self, reason: Failure | BaseException | None) -> None: ... class ReceivedMessage(NamedTuple): - channel: Incomplete - method: Incomplete - properties: Incomplete - body: Incomplete + channel: TwistedChannel + method: amqp_object.Method + properties: BasicProperties + body: bytes class TwistedChannel: on_closed: Deferred[Incomplete | Failure | BaseException | None] - def __init__(self, channel) -> None: ... + def __init__(self, channel: Channel) -> None: ... @property - def channel_number(self): ... + def channel_number(self) -> int: ... @property - def connection(self): ... + def connection(self) -> Connection: ... @property - def is_closed(self): ... + def is_closed(self) -> bool: ... @property - def is_closing(self): ... + def is_closing(self) -> bool: ... @property - def is_open(self): ... + def is_open(self) -> bool: ... @property - def flow_active(self): ... + def flow_active(self) -> bool: ... @property - def consumer_tags(self): ... - def callback_deferred(self, deferred, replies) -> None: ... - def add_on_return_callback(self, callback): ... - def basic_ack(self, delivery_tag: int = ..., multiple: bool = ...): ... - def basic_cancel(self, consumer_tag: str = ...) -> Deferred[Incomplete | Failure | BaseException | None]: ... + def consumer_tags(self) -> list[str]: ... + def callback_deferred(self, deferred: Deferred[Incomplete], replies: Iterable[Incomplete]) -> None: ... + def add_on_return_callback(self, callback: Callable[[ReceivedMessage], None]) -> None: ... + def basic_ack(self, delivery_tag: int = 0, multiple: bool = False) -> None: ... + def basic_cancel(self, consumer_tag: str = "") -> Deferred[Incomplete | Failure | BaseException | None]: ... def basic_consume( self, - queue, - auto_ack: bool = ..., - exclusive: bool = ..., - consumer_tag: Incomplete | None = ..., - arguments: Incomplete | None = ..., + queue: str, + auto_ack: bool = False, + exclusive: bool = False, + consumer_tag: str | None = None, + arguments: Mapping[str, Incomplete] | None = None, ) -> Deferred[Incomplete | Failure | BaseException]: ... - def basic_get(self, queue, auto_ack: bool = ...) -> Deferred[Incomplete | Failure | BaseException | None]: ... - def basic_nack(self, delivery_tag: Incomplete | None = ..., multiple: bool = ..., requeue: bool = ...): ... + def basic_get(self, queue: str, auto_ack: bool = False) -> Deferred[Incomplete | Failure | BaseException | None]: ... + def basic_nack(self, delivery_tag: int = 0, multiple: bool = False, requeue: bool = True) -> None: ... def basic_publish( - self, exchange, routing_key, body, properties: Incomplete | None = ..., mandatory: bool = ... + self, + exchange: str, + routing_key: str, + body: str | bytes, + properties: BasicProperties | None = None, + mandatory: bool = False, ) -> Deferred[Incomplete | Failure | BaseException]: ... def basic_qos( - self, prefetch_size: int = ..., prefetch_count: int = ..., global_qos: bool = ... + self, prefetch_size: int = 0, prefetch_count: int = 0, global_qos: bool = False ) -> Deferred[Incomplete | Failure | BaseException | None]: ... - def basic_reject(self, delivery_tag, requeue: bool = ...): ... - def basic_recover(self, requeue: bool = ...) -> Deferred[Incomplete | Failure | BaseException | None]: ... - def close(self, reply_code: int = ..., reply_text: str = ...): ... + def basic_reject(self, delivery_tag: int, requeue: bool = True): ... + def basic_recover(self, requeue: bool = False) -> Deferred[Incomplete | Failure | BaseException | None]: ... + def close(self, reply_code: int = 0, reply_text: str = "Normal shutdown"): ... def confirm_delivery(self) -> Deferred[Incomplete | None]: ... def exchange_bind( - self, destination, source, routing_key: str = ..., arguments: Incomplete | None = ... + self, destination: str, source: str, routing_key: str = "", arguments: Mapping[str, Incomplete] | None = None ) -> Deferred[Incomplete | Failure | BaseException | None]: ... def exchange_declare( self, exchange, - exchange_type=..., - passive: bool = ..., - durable: bool = ..., - auto_delete: bool = ..., - internal: bool = ..., - arguments: Incomplete | None = ..., + exchange_type: str | ExchangeType = ..., + passive: bool = False, + durable: bool = False, + auto_delete: bool = False, + internal: bool = False, + arguments: Mapping[str, Incomplete] | None = None, ) -> Deferred[Incomplete | Failure | BaseException | None]: ... def exchange_delete( - self, exchange: Incomplete | None = ..., if_unused: bool = ... + self, exchange: str | None = None, if_unused: bool = False ) -> Deferred[Incomplete | Failure | BaseException | None]: ... def exchange_unbind( - self, destination: str, source: str, routing_key: str = "", arguments: dict[str, Any] | None = None + self, destination: str, source: str, routing_key: str = "", arguments: Mapping[str, Incomplete] | None = None ) -> Deferred[Incomplete | Failure | BaseException | None]: ... def flow(self, active: bool = True) -> Deferred[Incomplete | Failure | BaseException | None]: ... def open(self): ... def queue_bind( - self, queue, exchange, routing_key: Incomplete | None = ..., arguments: Incomplete | None = ... + self, queue: str, exchange: str, routing_key: str | None = None, arguments: Mapping[str, Incomplete] | None = None ) -> Deferred[Incomplete | Failure | BaseException | None]: ... def queue_declare( self, - queue, - passive: bool = ..., - durable: bool = ..., - exclusive: bool = ..., - auto_delete: bool = ..., - arguments: Incomplete | None = ..., + queue: str, + passive: bool = False, + durable: bool = False, + exclusive: bool = False, + auto_delete: bool = False, + arguments: Mapping[str, Incomplete] | None = None, ) -> Deferred[Incomplete | Failure | BaseException | None]: ... def queue_delete( - self, queue, if_unused: bool = ..., if_empty: bool = ... + self, queue: str, if_unused: bool = False, if_empty: bool = False ) -> Deferred[Incomplete | Failure | BaseException | None]: ... - def queue_purge(self, queue) -> Deferred[Incomplete | Failure | BaseException | None]: ... + def queue_purge(self, queue: str) -> Deferred[Incomplete | Failure | BaseException | None]: ... def queue_unbind( - self, queue: str, exchange: str | None, routing_key: str | None = None, arguments: dict[str, Any] | None = None + self, queue: str, exchange: str | None, routing_key: str | None = None, arguments: Mapping[str, Incomplete] | None = None ) -> Deferred[Incomplete | Failure | BaseException | None]: ... def tx_commit(self) -> Deferred[Incomplete | Failure | BaseException | None]: ... def tx_rollback(self) -> Deferred[Incomplete | Failure | BaseException | None]: ... @@ -129,26 +140,28 @@ class _TwistedConnectionAdapter(Connection): on_open_callback: Callable[[Connection], object] | None, on_open_error_callback: Callable[[Connection, Exception], object] | None, on_close_callback: Callable[[Connection, Exception], object] | None, - custom_reactor=None, + custom_reactor: ReactorBase | None = None, ) -> None: ... def connection_made(self, transport: ITransport) -> None: ... def connection_lost(self, error: Exception) -> None: ... - def data_received(self, data) -> None: ... + def data_received(self, data: bytes) -> None: ... class TwistedProtocolConnection(Protocol): # pyright: ignore[reportUntypedBaseClass] ready: Deferred[None] | None closed: Deferred[None] | Failure | BaseException | None - def __init__(self, parameters: Incomplete | None = ..., custom_reactor: Incomplete | None = ...) -> None: ... - def channel(self, channel_number: Incomplete | None = ...): ... + def __init__(self, parameters: ConnectionParameters | None = None, custom_reactor: ReactorBase | None = None) -> None: ... + def channel(self, channel_number: int | None = None) -> Deferred[TwistedChannel]: ... @property - def is_open(self): ... + def is_open(self) -> bool: ... @property - def is_closed(self): ... - def close(self, reply_code: int = ..., reply_text: str = ...) -> Deferred[None] | Failure | BaseException | None: ... - def dataReceived(self, data) -> None: ... + def is_closed(self) -> bool: ... + def close( + self, reply_code: int = 200, reply_text: str = "Normal shutdown" + ) -> Deferred[None] | Failure | BaseException | None: ... + def dataReceived(self, data: bytes) -> None: ... def connectionLost(self, reason: Failure | BaseException = ...) -> None: ... def makeConnection(self, transport: ITransport) -> None: ... - def connectionReady(self): ... + def connectionReady(self) -> TwistedProtocolConnection | Deferred[TwistedProtocolConnection]: ... class _TimerHandle(AbstractTimerReference): def __init__(self, handle: DelayedCall) -> None: ... diff --git a/stubs/pika/pika/adapters/utils/selector_ioloop_adapter.pyi b/stubs/pika/pika/adapters/utils/selector_ioloop_adapter.pyi index ae987ad316d1..020017eacaf4 100644 --- a/stubs/pika/pika/adapters/utils/selector_ioloop_adapter.pyi +++ b/stubs/pika/pika/adapters/utils/selector_ioloop_adapter.pyi @@ -11,7 +11,7 @@ _Timeout = TypeVar("_Timeout", bound=object, default=object) @type_check_only class _SupportsCancel(Protocol): - def cancel(self): ... + def cancel(self) -> bool: ... class AbstractSelectorIOLoop(Generic[_Timeout], metaclass=abc.ABCMeta): @property