Skip to content

Commit 2e59b81

Browse files
AvasamAlexWaygood
Avasam
and
AlexWaygood
authored
Use of twisted in pika (#9494)
Co-authored-by: Alex Waygood <[email protected]>
1 parent 65f1884 commit 2e59b81

File tree

3 files changed

+66
-51
lines changed

3 files changed

+66
-51
lines changed

‎stubs/pika/@tests/stubtest_allowlist.txt

+5-12
Original file line numberDiff line numberDiff line change
@@ -3,18 +3,11 @@ pika.compat.StringIO.seek
33
pika.compat.StringIO.truncate
44

55
# Stubtest doesn't understand that a property alias is also read-only.
6-
pika.BlockingConnection.basic_nack
7-
pika.BlockingConnection.consumer_cancel_notify
8-
pika.BlockingConnection.exchange_exchange_bindings
9-
pika.BlockingConnection.publisher_confirms
10-
pika.adapters.BlockingConnection.basic_nack
11-
pika.adapters.BlockingConnection.consumer_cancel_notify
12-
pika.adapters.BlockingConnection.exchange_exchange_bindings
13-
pika.adapters.BlockingConnection.publisher_confirms
14-
pika.adapters.blocking_connection.BlockingConnection.basic_nack
15-
pika.adapters.blocking_connection.BlockingConnection.consumer_cancel_notify
16-
pika.adapters.blocking_connection.BlockingConnection.exchange_exchange_bindings
17-
pika.adapters.blocking_connection.BlockingConnection.publisher_confirms
6+
# https://github.com/python/mypy/issues/13975
7+
pika(\.adapters)?(\.blocking_connection)?\.BlockingConnection\.basic_nack
8+
pika(\.adapters)?(\.blocking_connection)?\.BlockingConnection\.consumer_cancel_notify
9+
pika(\.adapters)?(\.blocking_connection)?\.BlockingConnection\.exchange_exchange_bindings
10+
pika(\.adapters)?(\.blocking_connection)?\.BlockingConnection\.publisher_confirms
1811

1912
# The implementation has defaults for the arguments that would make the
2013
# created instances unusable, so we require the arguments in the stub.

‎stubs/pika/pika/adapters/twisted_connection.pyi

+58-39
Original file line numberDiff line numberDiff line change
@@ -1,22 +1,29 @@
1+
# twisted is optional and self-contained in this module.
2+
# 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.
3+
14
from _typeshed import Incomplete
2-
from typing import Any, NamedTuple
3-
from typing_extensions import TypeAlias
5+
from typing import NamedTuple, TypeVar
46

57
import pika.connection
68
from pika.adapters.utils import nbio_interface
9+
from twisted.internet.base import DelayedCall # type: ignore[import] # pyright: ignore[reportMissingImports]
10+
from twisted.internet.defer import Deferred, DeferredQueue # type: ignore[import] # pyright: ignore[reportMissingImports]
11+
from twisted.internet.interfaces import ITransport # type: ignore[import] # pyright: ignore[reportMissingImports]
12+
from twisted.internet.protocol import Protocol # type: ignore[import] # pyright: ignore[reportMissingImports]
13+
from twisted.python.failure import Failure # type: ignore[import] # pyright: ignore[reportMissingImports]
714

8-
_DeferredQueue: TypeAlias = Any # TODO: twisted.internet.defer.DeferredQueue
9-
_Protocol: TypeAlias = Any # TODO: twisted.internet.protocol.Protocol
15+
_T = TypeVar("_T")
1016

1117
LOGGER: Incomplete
1218

13-
class ClosableDeferredQueue(_DeferredQueue):
14-
closed: Incomplete
19+
class ClosableDeferredQueue(DeferredQueue[_T]): # pyright: ignore[reportUntypedBaseClass]
20+
closed: Failure | BaseException | None
1521
def __init__(self, size: Incomplete | None = ..., backlog: Incomplete | None = ...) -> None: ...
16-
def put(self, obj): ...
17-
def get(self): ...
22+
# Returns a Deferred with an error if fails. None if success
23+
def put(self, obj: _T) -> Deferred[Failure | BaseException] | None: ... # type: ignore[override] # pyright: ignore[reportInvalidTypeVarUse]
24+
def get(self) -> Deferred[Failure | BaseException | _T]: ...
1825
pending: Incomplete
19-
def close(self, reason) -> None: ...
26+
def close(self, reason: BaseException | None) -> None: ...
2027

2128
class ReceivedMessage(NamedTuple):
2229
channel: Incomplete
@@ -25,7 +32,7 @@ class ReceivedMessage(NamedTuple):
2532
body: Incomplete
2633

2734
class TwistedChannel:
28-
on_closed: Incomplete
35+
on_closed: Deferred[Incomplete | Failure | BaseException | None]
2936
def __init__(self, channel) -> None: ...
3037
@property
3138
def channel_number(self): ...
@@ -44,24 +51,30 @@ class TwistedChannel:
4451
def callback_deferred(self, deferred, replies) -> None: ...
4552
def add_on_return_callback(self, callback): ...
4653
def basic_ack(self, delivery_tag: int = ..., multiple: bool = ...): ...
47-
def basic_cancel(self, consumer_tag: str = ...): ...
54+
def basic_cancel(self, consumer_tag: str = ...) -> Deferred[Incomplete | Failure | BaseException | None]: ...
4855
def basic_consume(
4956
self,
5057
queue,
5158
auto_ack: bool = ...,
5259
exclusive: bool = ...,
5360
consumer_tag: Incomplete | None = ...,
5461
arguments: Incomplete | None = ...,
55-
): ...
56-
def basic_get(self, queue, auto_ack: bool = ...): ...
62+
) -> Deferred[Incomplete | Failure | BaseException]: ...
63+
def basic_get(self, queue, auto_ack: bool = ...) -> Deferred[Incomplete | Failure | BaseException | None]: ...
5764
def basic_nack(self, delivery_tag: Incomplete | None = ..., multiple: bool = ..., requeue: bool = ...): ...
58-
def basic_publish(self, exchange, routing_key, body, properties: Incomplete | None = ..., mandatory: bool = ...): ...
59-
def basic_qos(self, prefetch_size: int = ..., prefetch_count: int = ..., global_qos: bool = ...): ...
65+
def basic_publish(
66+
self, exchange, routing_key, body, properties: Incomplete | None = ..., mandatory: bool = ...
67+
) -> Deferred[Incomplete | Failure | BaseException]: ...
68+
def basic_qos(
69+
self, prefetch_size: int = ..., prefetch_count: int = ..., global_qos: bool = ...
70+
) -> Deferred[Incomplete | Failure | BaseException | None]: ...
6071
def basic_reject(self, delivery_tag, requeue: bool = ...): ...
61-
def basic_recover(self, requeue: bool = ...): ...
72+
def basic_recover(self, requeue: bool = ...) -> Deferred[Incomplete | Failure | BaseException | None]: ...
6273
def close(self, reply_code: int = ..., reply_text: str = ...): ...
63-
def confirm_delivery(self): ...
64-
def exchange_bind(self, destination, source, routing_key: str = ..., arguments: Incomplete | None = ...): ...
74+
def confirm_delivery(self) -> Deferred[Incomplete | None]: ...
75+
def exchange_bind(
76+
self, destination, source, routing_key: str = ..., arguments: Incomplete | None = ...
77+
) -> Deferred[Incomplete | Failure | BaseException | None]: ...
6578
def exchange_declare(
6679
self,
6780
exchange,
@@ -71,18 +84,22 @@ class TwistedChannel:
7184
auto_delete: bool = ...,
7285
internal: bool = ...,
7386
arguments: Incomplete | None = ...,
74-
): ...
75-
def exchange_delete(self, exchange: Incomplete | None = ..., if_unused: bool = ...): ...
87+
) -> Deferred[Incomplete | Failure | BaseException | None]: ...
88+
def exchange_delete(
89+
self, exchange: Incomplete | None = ..., if_unused: bool = ...
90+
) -> Deferred[Incomplete | Failure | BaseException | None]: ...
7691
def exchange_unbind(
7792
self,
7893
destination: Incomplete | None = ...,
7994
source: Incomplete | None = ...,
8095
routing_key: str = ...,
8196
arguments: Incomplete | None = ...,
82-
): ...
83-
def flow(self, active): ...
97+
) -> Deferred[Incomplete | Failure | BaseException | None]: ...
98+
def flow(self, active) -> Deferred[Incomplete | Failure | BaseException | None]: ...
8499
def open(self): ...
85-
def queue_bind(self, queue, exchange, routing_key: Incomplete | None = ..., arguments: Incomplete | None = ...): ...
100+
def queue_bind(
101+
self, queue, exchange, routing_key: Incomplete | None = ..., arguments: Incomplete | None = ...
102+
) -> Deferred[Incomplete | Failure | BaseException | None]: ...
86103
def queue_declare(
87104
self,
88105
queue,
@@ -91,37 +108,39 @@ class TwistedChannel:
91108
exclusive: bool = ...,
92109
auto_delete: bool = ...,
93110
arguments: Incomplete | None = ...,
94-
): ...
95-
def queue_delete(self, queue, if_unused: bool = ..., if_empty: bool = ...): ...
96-
def queue_purge(self, queue): ...
111+
) -> Deferred[Incomplete | Failure | BaseException | None]: ...
112+
def queue_delete(
113+
self, queue, if_unused: bool = ..., if_empty: bool = ...
114+
) -> Deferred[Incomplete | Failure | BaseException | None]: ...
115+
def queue_purge(self, queue) -> Deferred[Incomplete | Failure | BaseException | None]: ...
97116
def queue_unbind(
98117
self, queue, exchange: Incomplete | None = ..., routing_key: Incomplete | None = ..., arguments: Incomplete | None = ...
99-
): ...
100-
def tx_commit(self): ...
101-
def tx_rollback(self): ...
102-
def tx_select(self): ...
118+
) -> Deferred[Incomplete | Failure | BaseException | None]: ...
119+
def tx_commit(self) -> Deferred[Incomplete | Failure | BaseException | None]: ...
120+
def tx_rollback(self) -> Deferred[Incomplete | Failure | BaseException | None]: ...
121+
def tx_select(self) -> Deferred[Incomplete | Failure | BaseException | None]: ...
103122

104123
class _TwistedConnectionAdapter(pika.connection.Connection):
105124
def __init__(self, parameters, on_open_callback, on_open_error_callback, on_close_callback, custom_reactor) -> None: ...
106-
def connection_made(self, transport) -> None: ...
107-
def connection_lost(self, error) -> None: ...
125+
def connection_made(self, transport: ITransport) -> None: ...
126+
def connection_lost(self, error: Exception) -> None: ...
108127
def data_received(self, data) -> None: ...
109128

110-
class TwistedProtocolConnection(_Protocol):
111-
ready: Incomplete
112-
closed: Incomplete
129+
class TwistedProtocolConnection(Protocol): # pyright: ignore[reportUntypedBaseClass]
130+
ready: Deferred[None] | None
131+
closed: Deferred[None] | Failure | BaseException | None
113132
def __init__(self, parameters: Incomplete | None = ..., custom_reactor: Incomplete | None = ...) -> None: ...
114133
def channel(self, channel_number: Incomplete | None = ...): ...
115134
@property
116135
def is_open(self): ...
117136
@property
118137
def is_closed(self): ...
119-
def close(self, reply_code: int = ..., reply_text: str = ...): ...
138+
def close(self, reply_code: int = ..., reply_text: str = ...) -> Deferred[None] | Failure | BaseException | None: ...
120139
def dataReceived(self, data) -> None: ...
121-
def connectionLost(self, reason=...) -> None: ...
122-
def makeConnection(self, transport) -> None: ...
140+
def connectionLost(self, reason: Failure | BaseException = ...) -> None: ...
141+
def makeConnection(self, transport: ITransport) -> None: ...
123142
def connectionReady(self): ...
124143

125144
class _TimerHandle(nbio_interface.AbstractTimerReference):
126-
def __init__(self, handle) -> None: ...
145+
def __init__(self, handle: DelayedCall) -> None: ...
127146
def cancel(self) -> None: ...

‎tests/pytype_exclude_list.txt

+3
Original file line numberDiff line numberDiff line change
@@ -10,6 +10,9 @@ stubs/mysqlclient/MySQLdb/__init__.pyi
1010
stubs/mysqlclient/MySQLdb/connections.pyi
1111
stubs/mysqlclient/MySQLdb/cursors.pyi
1212

13+
# twisted not installed during tests
14+
stubs/pika/pika/adapters/twisted_connection.pyi
15+
1316
# _pb2.pyi have some constructs that break pytype
1417
# Eg
1518
# pytype.pyi.parser.ParseError: File: "/Users/nipunn/src/typeshed/third_party/2and3/google/protobuf/descriptor_pb2.pyi", line 195

0 commit comments

Comments
 (0)