Initial pika stubs (#9200)

Author: srittau

pyrightconfig.stricter.json

@@ -52,6 +52,7 @@
         "stubs/google-cloud-ndb",
         "stubs/passlib",
         "stubs/peewee",
+        "stubs/pika",
         "stubs/psutil",
         "stubs/psycopg2",
         "stubs/pyflakes",

stubs/pika/@tests/stubtest_allowlist.txt

@@ -0,0 +1,28 @@
+# Inherited from stdlib's io module.
+pika.compat.StringIO.seek
+pika.compat.StringIO.truncate
+
+# Requires external libraries to be installed.
+pika\.adapters\.gevent_connection.*
+pika\.adapters\.tornado_connection.*
+pika\.adapters\.twisted_connection.*
+
+# Stubtest doesn't understand that a property alias is also read-only.
+pika.BlockingConnection.basic_nack
+pika.BlockingConnection.consumer_cancel_notify
+pika.BlockingConnection.exchange_exchange_bindings
+pika.BlockingConnection.publisher_confirms
+pika.adapters.BlockingConnection.basic_nack
+pika.adapters.BlockingConnection.consumer_cancel_notify
+pika.adapters.BlockingConnection.exchange_exchange_bindings
+pika.adapters.BlockingConnection.publisher_confirms
+pika.adapters.blocking_connection.BlockingConnection.basic_nack
+pika.adapters.blocking_connection.BlockingConnection.consumer_cancel_notify
+pika.adapters.blocking_connection.BlockingConnection.exchange_exchange_bindings
+pika.adapters.blocking_connection.BlockingConnection.publisher_confirms
+
+# Flagged by stubtest for unknown reasons.
+pika.data.PY2
+pika.data.basestring
+pika.spec.str_or_bytes
+pika.validators.basestring

stubs/pika/METADATA.toml

@@ -0,0 +1,4 @@
+version = "1.3.*"
+
+[tool.stubtest]
+ignore_missing_stub = false

stubs/pika/pika/__init__.pyi

@@ -0,0 +1,11 @@
+from pika import adapters as adapters
+from pika.adapters import (
+    BaseConnection as BaseConnection,
+    BlockingConnection as BlockingConnection,
+    SelectConnection as SelectConnection,
+)
+from pika.adapters.utils.connection_workflow import AMQPConnectionWorkflow as AMQPConnectionWorkflow
+from pika.connection import ConnectionParameters as ConnectionParameters, SSLOptions as SSLOptions, URLParameters as URLParameters
+from pika.credentials import PlainCredentials as PlainCredentials
+from pika.delivery_mode import DeliveryMode as DeliveryMode
+from pika.spec import BasicProperties as BasicProperties

stubs/pika/pika/adapters/__init__.pyi

@@ -0,0 +1,3 @@
+from pika.adapters.base_connection import BaseConnection as BaseConnection
+from pika.adapters.blocking_connection import BlockingConnection as BlockingConnection
+from pika.adapters.select_connection import IOLoop as IOLoop, SelectConnection as SelectConnection

stubs/pika/pika/adapters/asyncio_connection.pyi

@@ -0,0 +1,48 @@
+from _typeshed import Incomplete
+
+from pika.adapters import base_connection
+from pika.adapters.utils import io_services_utils, nbio_interface
+
+LOGGER: Incomplete
+
+class AsyncioConnection(base_connection.BaseConnection):
+    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 = ...,
+    ) -> None: ...
+    @classmethod
+    def create_connection(
+        cls, connection_configs, on_done, custom_ioloop: Incomplete | None = ..., workflow: Incomplete | None = ...
+    ): ...
+
+class _AsyncioIOServicesAdapter(
+    io_services_utils.SocketConnectionMixin,
+    io_services_utils.StreamingConnectionMixin,
+    nbio_interface.AbstractIOServices,
+    nbio_interface.AbstractFileDescriptorServices,
+):
+    def __init__(self, loop: Incomplete | None = ...) -> None: ...
+    def get_native_ioloop(self): ...
+    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 = ..., socktype: int = ..., proto: int = ..., flags: int = ...): ...
+    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): ...
+
+class _TimerHandle(nbio_interface.AbstractTimerReference):
+    def __init__(self, handle) -> None: ...
+    def cancel(self) -> None: ...
+
+class _AsyncioIOReference(nbio_interface.AbstractIOReference):
+    def __init__(self, future, on_done) -> None: ...
+    def cancel(self): ...

stubs/pika/pika/adapters/base_connection.pyi

@@ -0,0 +1,28 @@
+import abc
+from _typeshed import Incomplete
+
+from pika import connection
+from pika.adapters.utils import nbio_interface
+
+LOGGER: Incomplete
+
+class BaseConnection(connection.Connection, metaclass=abc.ABCMeta):
+    def __init__(
+        self, parameters, on_open_callback, on_open_error_callback, on_close_callback, nbio, internal_connection_workflow
+    ) -> None: ...
+    @classmethod
+    @abc.abstractmethod
+    def create_connection(
+        cls, connection_configs, on_done, custom_ioloop: Incomplete | None = ..., workflow: Incomplete | None = ...
+    ): ...
+    @property
+    def ioloop(self): ...
+
+class _StreamingProtocolShim(nbio_interface.AbstractStreamProtocol):
+    connection_made: Incomplete
+    connection_lost: Incomplete
+    eof_received: Incomplete
+    data_received: Incomplete
+    conn: Incomplete
+    def __init__(self, conn) -> None: ...
+    def __getattr__(self, attr): ...

stubs/pika/pika/adapters/blocking_connection.pyi

@@ -0,0 +1,236 @@
+from _typeshed import Incomplete
+from collections.abc import Generator
+from typing import NamedTuple
+
+LOGGER: Incomplete
+
+class _CallbackResult:
+    def __init__(self, value_class: Incomplete | None = ...) -> None: ...
+    def reset(self) -> None: ...
+    def __bool__(self) -> bool: ...
+    __nonzero__: Incomplete
+    def __enter__(self): ...
+    def __exit__(self, *args, **kwargs) -> 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, **_kwargs) -> None: ...
+    def is_ready(self): ...
+
+class _TimerEvt:
+    timer_id: Incomplete
+    def __init__(self, callback) -> None: ...
+    def dispatch(self) -> None: ...
+
+class _ConnectionBlockedUnblockedEvtBase:
+    def __init__(self, callback, method_frame) -> None: ...
+    def dispatch(self) -> None: ...
+
+class _ConnectionBlockedEvt(_ConnectionBlockedUnblockedEvtBase): ...
+class _ConnectionUnblockedEvt(_ConnectionBlockedUnblockedEvtBase): ...
+
+class BlockingConnection:
+    class _OnClosedArgs(NamedTuple):
+        connection: Incomplete
+        error: Incomplete
+
+    class _OnChannelOpenedArgs(NamedTuple):
+        channel: Incomplete
+    def __init__(self, parameters: Incomplete | None = ..., _impl_class: Incomplete | None = ...) -> None: ...
+    def __enter__(self): ...
+    def __exit__(self, exc_type, value, traceback) -> 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 close(self, reply_code: int = ..., reply_text: str = ...) -> None: ...
+    def process_data_events(self, time_limit: int = ...): ...
+    def sleep(self, duration) -> None: ...
+    def channel(self, channel_number: Incomplete | None = ...): ...
+    @property
+    def is_closed(self): ...
+    @property
+    def is_open(self): ...
+    @property
+    def basic_nack_supported(self): ...
+    @property
+    def consumer_cancel_notify_supported(self): ...
+    @property
+    def exchange_exchange_bindings_supported(self): ...
+    @property
+    def publisher_confirms_supported(self): ...
+    basic_nack = basic_nack_supported
+    consumer_cancel_notify = consumer_cancel_notify_supported
+    exchange_exchange_bindings = exchange_exchange_bindings_supported
+    publisher_confirms = publisher_confirms_supported
+
+class _ChannelPendingEvt: ...
+
+class _ConsumerDeliveryEvt(_ChannelPendingEvt):
+    method: Incomplete
+    properties: Incomplete
+    body: Incomplete
+    def __init__(self, method, properties, body) -> None: ...
+
+class _ConsumerCancellationEvt(_ChannelPendingEvt):
+    method_frame: Incomplete
+    def __init__(self, method_frame) -> None: ...
+    @property
+    def method(self): ...
+
+class _ReturnedMessageEvt(_ChannelPendingEvt):
+    callback: Incomplete
+    channel: Incomplete
+    method: Incomplete
+    properties: Incomplete
+    body: Incomplete
+    def __init__(self, callback, channel, method, properties, body) -> None: ...
+    def dispatch(self) -> None: ...
+
+class ReturnedMessage:
+    method: Incomplete
+    properties: Incomplete
+    body: Incomplete
+    def __init__(self, method, properties, body) -> None: ...
+
+class _ConsumerInfo:
+    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: Incomplete | None = ..., alternate_event_sink: Incomplete | None = ...
+    ) -> None: ...
+    @property
+    def setting_up(self): ...
+    @property
+    def active(self): ...
+    @property
+    def tearing_down(self): ...
+    @property
+    def cancelled_by_broker(self): ...
+
+class _QueueConsumerGeneratorInfo:
+    params: Incomplete
+    consumer_tag: Incomplete
+    pending_events: Incomplete
+    def __init__(self, params, consumer_tag) -> 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 __int__(self) -> int: ...
+    def __enter__(self): ...
+    def __exit__(self, exc_type, value, traceback) -> None: ...
+    @property
+    def channel_number(self): ...
+    @property
+    def connection(self): ...
+    @property
+    def is_closed(self): ...
+    @property
+    def is_open(self): ...
+    @property
+    def consumer_tags(self): ...
+    def close(self, reply_code: int = ..., reply_text: str = ...): ...
+    def flow(self, active): ...
+    def add_on_cancel_callback(self, callback) -> None: ...
+    def add_on_return_callback(self, callback): ...
+    def basic_consume(
+        self,
+        queue,
+        on_message_callback,
+        auto_ack: bool = ...,
+        exclusive: bool = ...,
+        consumer_tag: Incomplete | None = ...,
+        arguments: Incomplete | None = ...,
+    ): ...
+    def basic_cancel(self, consumer_tag): ...
+    def start_consuming(self) -> None: ...
+    def stop_consuming(self, consumer_tag: Incomplete | None = ...) -> None: ...
+    def consume(
+        self,
+        queue,
+        auto_ack: bool = ...,
+        exclusive: bool = ...,
+        arguments: Incomplete | None = ...,
+        inactivity_timeout: Incomplete | None = ...,
+    ) -> Generator[Incomplete, None, None]: ...
+    def get_waiting_message_count(self): ...
+    def cancel(self): ...
+    def basic_ack(self, delivery_tag: int = ..., multiple: bool = ...) -> None: ...
+    def basic_nack(self, delivery_tag: int = ..., multiple: bool = ..., requeue: bool = ...) -> None: ...
+    def basic_get(self, queue, auto_ack: bool = ...): ...
+    def basic_publish(self, exchange, routing_key, body, properties: Incomplete | None = ..., mandatory: bool = ...) -> None: ...
+    def basic_qos(self, prefetch_size: int = ..., prefetch_count: int = ..., global_qos: bool = ...) -> None: ...
+    def basic_recover(self, requeue: bool = ...) -> None: ...
+    def basic_reject(self, delivery_tag: int = ..., requeue: bool = ...) -> None: ...
+    def confirm_delivery(self) -> None: ...
+    def exchange_declare(
+        self,
+        exchange,
+        exchange_type=...,
+        passive: bool = ...,
+        durable: bool = ...,
+        auto_delete: bool = ...,
+        internal: bool = ...,
+        arguments: Incomplete | None = ...,
+    ): ...
+    def exchange_delete(self, exchange: Incomplete | None = ..., if_unused: bool = ...): ...
+    def exchange_bind(self, destination, source, routing_key: str = ..., arguments: Incomplete | None = ...): ...
+    def exchange_unbind(
+        self,
+        destination: Incomplete | None = ...,
+        source: Incomplete | None = ...,
+        routing_key: str = ...,
+        arguments: Incomplete | None = ...,
+    ): ...
+    def queue_declare(
+        self,
+        queue,
+        passive: bool = ...,
+        durable: bool = ...,
+        exclusive: bool = ...,
+        auto_delete: bool = ...,
+        arguments: Incomplete | None = ...,
+    ): ...
+    def queue_delete(self, queue, if_unused: bool = ..., if_empty: bool = ...): ...
+    def queue_purge(self, queue): ...
+    def queue_bind(self, queue, exchange, routing_key: Incomplete | None = ..., arguments: Incomplete | None = ...): ...
+    def queue_unbind(
+        self, queue, exchange: Incomplete | None = ..., routing_key: Incomplete | None = ..., arguments: Incomplete | None = ...
+    ): ...
+    def tx_select(self): ...
+    def tx_commit(self): ...
+    def tx_rollback(self): ...