Skip to content

Commit bc600bb

Browse files
plamutplamut
authored
fix: refactor client classes for safer type checking (#552)
* Directly subclass generated PublisherClient * Directly subclass generated SubscriberClient * Remove unneded GAPIC helper * Remove pytype There are no real advantages over mypy, but at the same time several downsides such as being slow, producing more false positives, etc. * Re-enable mypy_samples nox session * Convert a comment to docstring in publisher client * Add api property back, but deprecated * Assure that mypy_samples is not commented out * Remove redundant type hint casts * Disable mypy_samples session until blockers resolved
1 parent 779fc16 commit bc600bb

File tree

19 files changed

+180
-325
lines changed

19 files changed

+180
-325
lines changed

‎packages/google-cloud-pubsub/.gitignore‎

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -29,7 +29,6 @@ pip-log.txt
2929
.nox
3030
.cache
3131
.pytest_cache
32-
.pytype
3332

3433

3534
# Mac

‎packages/google-cloud-pubsub/google/cloud/pubsub_v1/_gapic.py‎

Lines changed: 0 additions & 74 deletions
This file was deleted.

‎packages/google-cloud-pubsub/google/cloud/pubsub_v1/publisher/_batch/thread.py‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -271,7 +271,7 @@ def _commit(self) -> None:
271271
batch_transport_succeeded = True
272272
try:
273273
# Performs retries for errors defined by the retry configuration.
274-
response = self._client.api.publish(
274+
response = self._client._gapic_publish(
275275
topic=self._topic,
276276
messages=self._messages,
277277
retry=self._commit_retry,

‎packages/google-cloud-pubsub/google/cloud/pubsub_v1/publisher/client.py‎

Lines changed: 33 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -22,12 +22,12 @@
2222
import time
2323
import typing
2424
from typing import Any, Dict, Optional, Sequence, Tuple, Type, Union
25+
import warnings
2526

2627
from google.api_core import gapic_v1
2728
from google.auth.credentials import AnonymousCredentials # type: ignore
2829
from google.oauth2 import service_account # type: ignore
2930

30-
from google.cloud.pubsub_v1 import _gapic
3131
from google.cloud.pubsub_v1 import types
3232
from google.cloud.pubsub_v1.publisher import exceptions
3333
from google.cloud.pubsub_v1.publisher import futures
@@ -49,15 +49,11 @@
4949
from google.cloud import pubsub_v1
5050
from google.cloud.pubsub_v1.publisher import _batch
5151
from google.pubsub_v1.services.publisher.client import OptionalRetry
52+
from google.pubsub_v1.types import pubsub as pubsub_types
5253

5354

5455
_LOGGER = logging.getLogger(__name__)
5556

56-
_DENYLISTED_METHODS = (
57-
"publish",
58-
"from_service_account_file",
59-
"from_service_account_json",
60-
)
6157

6258
_raw_proto_pubbsub_message = gapic_types.PubsubMessage.pb()
6359

@@ -66,8 +62,7 @@
6662
]
6763

6864

69-
@_gapic.add_methods(publisher_client.PublisherClient, denylist=_DENYLISTED_METHODS)
70-
class Client(object):
65+
class Client(publisher_client.PublisherClient):
7166
"""A publisher client for Google Cloud Pub/Sub.
7267
7368
This creates an object that is capable of publishing messages.
@@ -146,8 +141,8 @@ def __init__(
146141

147142
# Add the metrics headers, and instantiate the underlying GAPIC
148143
# client.
149-
self.api = publisher_client.PublisherClient(**kwargs)
150-
self._target = self.api._transport._host
144+
super().__init__(**kwargs)
145+
self._target = self._transport._host
151146
self._batch_class = thread.Batch
152147
self.batch_settings = types.BatchSettings(*batch_settings)
153148

@@ -164,7 +159,7 @@ def __init__(
164159
self._flow_controller = FlowController(self.publisher_options.flow_control)
165160

166161
@classmethod
167-
def from_service_account_file(
162+
def from_service_account_file( # type: ignore[override]
168163
cls,
169164
filename: str,
170165
batch_settings: Union[types.BatchSettings, Sequence] = (),
@@ -188,7 +183,7 @@ def from_service_account_file(
188183
kwargs["credentials"] = credentials
189184
return cls(batch_settings, **kwargs)
190185

191-
from_service_account_json = from_service_account_file
186+
from_service_account_json = from_service_account_file # type: ignore[assignment]
192187

193188
@property
194189
def target(self) -> str:
@@ -199,6 +194,26 @@ def target(self) -> str:
199194
"""
200195
return self._target
201196

197+
@property
198+
def api(self):
199+
"""The underlying gapic API client.
200+
201+
.. versionchanged:: 2.10.0
202+
Instead of a GAPIC ``PublisherClient`` client instance, this property is a
203+
proxy object to it with the same interface.
204+
205+
.. deprecated:: 2.10.0
206+
Use the GAPIC methods and properties on the client instance directly
207+
instead of through the :attr:`api` attribute.
208+
"""
209+
msg = (
210+
'The "api" property only exists for backward compatibility, access its '
211+
'attributes directly thorugh the client instance (e.g. "client.foo" '
212+
'instead of "client.api.foo").'
213+
)
214+
warnings.warn(msg, category=DeprecationWarning)
215+
return super()
216+
202217
def _get_or_create_sequencer(self, topic: str, ordering_key: str) -> SequencerType:
203218
""" Get an existing sequencer or create a new one given the (topic,
204219
ordering_key) pair.
@@ -252,7 +267,11 @@ def resume_publish(self, topic: str, ordering_key: str) -> None:
252267
else:
253268
sequencer.unpause()
254269

255-
def publish(
270+
def _gapic_publish(self, *args, **kwargs) -> "pubsub_types.PublishResponse":
271+
"""Call the GAPIC public API directly."""
272+
return super().publish(*args, **kwargs)
273+
274+
def publish( # type: ignore[override]
256275
self,
257276
topic: str,
258277
data: bytes,
@@ -382,7 +401,7 @@ def on_publish_done(future):
382401
if self._enable_message_ordering:
383402
if retry is gapic_v1.method.DEFAULT:
384403
# use the default retry for the publish GRPC method as a base
385-
transport = self.api._transport
404+
transport = self._transport
386405
base_retry = transport._wrapped_methods[transport.publish]._retry
387406
retry = base_retry.with_deadline(2.0 ** 32)
388407
else:

‎packages/google-cloud-pubsub/google/cloud/pubsub_v1/publisher/flow_controller.py‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -25,7 +25,7 @@
2525
_LOGGER = logging.getLogger(__name__)
2626

2727

28-
MessageType = Type[types.PubsubMessage] # type: ignore # pytype: disable=module-attr
28+
MessageType = Type[types.PubsubMessage] # type: ignore
2929

3030

3131
class _QuantityReservation:

‎packages/google-cloud-pubsub/google/cloud/pubsub_v1/subscriber/_protocol/leaser.py‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -76,7 +76,7 @@ def message_count(self) -> int:
7676
return len(self._leased_messages)
7777

7878
@property
79-
def ack_ids(self) -> KeysView[str]: # pytype: disable=invalid-annotation
79+
def ack_ids(self) -> KeysView[str]:
8080
"""The ack IDs of all leased messages."""
8181
return self._leased_messages.keys()
8282

‎packages/google-cloud-pubsub/google/cloud/pubsub_v1/subscriber/_protocol/streaming_pull_manager.py‎

Lines changed: 1 addition & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -532,7 +532,7 @@ def open(
532532
self._get_initial_request, stream_ack_deadline_seconds
533533
)
534534
self._rpc = bidi.ResumableBidiRpc(
535-
start_rpc=self._client.api.streaming_pull,
535+
start_rpc=self._client.streaming_pull,
536536
initial_request=get_initial_request,
537537
should_recover=self._should_recover,
538538
should_terminate=self._should_terminate,
@@ -548,14 +548,11 @@ def open(
548548

549549
# Create references to threads
550550
assert self._scheduler is not None
551-
# pytype: disable=wrong-arg-types
552-
# (pytype incorrectly complains about "self" not being the right argument type)
553551
scheduler_queue = self._scheduler.queue
554552
self._dispatcher = dispatcher.Dispatcher(self, scheduler_queue)
555553
self._consumer = bidi.BackgroundConsumer(self._rpc, self._on_response)
556554
self._leaser = leaser.Leaser(self)
557555
self._heartbeater = heartbeater.Heartbeater(self)
558-
# pytype: enable=wrong-arg-types
559556

560557
# Start the thread to pass the requests.
561558
self._dispatcher.start()

‎packages/google-cloud-pubsub/google/cloud/pubsub_v1/subscriber/client.py‎

Lines changed: 29 additions & 19 deletions
Original file line numberDiff line numberDiff line change
@@ -18,11 +18,11 @@
1818
import pkg_resources
1919
import typing
2020
from typing import cast, Any, Callable, Optional, Sequence, Union
21+
import warnings
2122

2223
from google.auth.credentials import AnonymousCredentials # type: ignore
2324
from google.oauth2 import service_account # type: ignore
2425

25-
from google.cloud.pubsub_v1 import _gapic
2626
from google.cloud.pubsub_v1 import types
2727
from google.cloud.pubsub_v1.subscriber import futures
2828
from google.cloud.pubsub_v1.subscriber._protocol import streaming_pull_manager
@@ -42,15 +42,8 @@
4242
# a PIP package.
4343
__version__ = "0.0"
4444

45-
_DENYLISTED_METHODS = (
46-
"publish",
47-
"from_service_account_file",
48-
"from_service_account_json",
49-
)
5045

51-
52-
@_gapic.add_methods(subscriber_client.SubscriberClient, denylist=_DENYLISTED_METHODS)
53-
class Client(object):
46+
class Client(subscriber_client.SubscriberClient):
5447
"""A subscriber client for Google Cloud Pub/Sub.
5548
5649
This creates an object that is capable of subscribing to messages.
@@ -91,12 +84,14 @@ def __init__(self, **kwargs: Any):
9184
kwargs["credentials"] = AnonymousCredentials()
9285

9386
# Instantiate the underlying GAPIC client.
94-
self._api = subscriber_client.SubscriberClient(**kwargs)
95-
self._target = self._api._transport._host
87+
super().__init__(**kwargs)
88+
self._target = self._transport._host
9689
self._closed = False
9790

9891
@classmethod
99-
def from_service_account_file(cls, filename: str, **kwargs: Any) -> "Client":
92+
def from_service_account_file( # type: ignore[override]
93+
cls, filename: str, **kwargs: Any
94+
) -> "Client":
10095
"""Creates an instance of this client using the provided credentials
10196
file.
10297
@@ -112,7 +107,7 @@ def from_service_account_file(cls, filename: str, **kwargs: Any) -> "Client":
112107
kwargs["credentials"] = credentials
113108
return cls(**kwargs)
114109

115-
from_service_account_json = from_service_account_file
110+
from_service_account_json = from_service_account_file # type: ignore[assignment]
116111

117112
@property
118113
def target(self) -> str:
@@ -123,11 +118,6 @@ def target(self) -> str:
123118
"""
124119
return self._target
125120

126-
@property
127-
def api(self) -> subscriber_client.SubscriberClient:
128-
"""The underlying gapic API client."""
129-
return self._api
130-
131121
@property
132122
def closed(self) -> bool:
133123
"""Return whether the client has been closed and cannot be used anymore.
@@ -136,6 +126,26 @@ def closed(self) -> bool:
136126
"""
137127
return self._closed
138128

129+
@property
130+
def api(self):
131+
"""The underlying gapic API client.
132+
133+
.. versionchanged:: 2.10.0
134+
Instead of a GAPIC ``SubscriberClient`` client instance, this property is a
135+
proxy object to it with the same interface.
136+
137+
.. deprecated:: 2.10.0
138+
Use the GAPIC methods and properties on the client instance directly
139+
instead of through the :attr:`api` attribute.
140+
"""
141+
msg = (
142+
'The "api" property only exists for backward compatibility, access its '
143+
'attributes directly thorugh the client instance (e.g. "client.foo" '
144+
'instead of "client.api.foo").'
145+
)
146+
warnings.warn(msg, category=DeprecationWarning)
147+
return super()
148+
139149
def subscribe(
140150
self,
141151
subscription: str,
@@ -266,7 +276,7 @@ def close(self) -> None:
266276
267277
This method is idempotent.
268278
"""
269-
transport = cast("SubscriberGrpcTransport", self.api._transport)
279+
transport = cast("SubscriberGrpcTransport", self._transport)
270280
transport.grpc_channel.close()
271281
self._closed = True
272282

‎packages/google-cloud-pubsub/google/cloud/pubsub_v1/subscriber/message.py‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -81,7 +81,7 @@ class Message(object):
8181
The time that this message was originally published.
8282
"""
8383

84-
def __init__( # pytype: disable=module-attr
84+
def __init__(
8585
self,
8686
message: "types.PubsubMessage._meta._pb", # type: ignore
8787
ack_id: str,

‎packages/google-cloud-pubsub/google/cloud/pubsub_v1/types.py‎

Lines changed: 0 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -127,13 +127,11 @@ class PublisherOptions(NamedTuple):
127127
"an instance of :class:`google.api_core.retry.Retry`."
128128
)
129129

130-
# pytype: disable=invalid-annotation
131130
timeout: "OptionalTimeout" = gapic_v1.method.DEFAULT # use api_core default
132131
(
133132
"Timeout settings for message publishing by the client. It should be "
134133
"compatible with :class:`~.pubsub_v1.types.TimeoutType`."
135134
)
136-
# pytype: enable=invalid-annotation
137135

138136

139137
# Define the type class and default values for flow control settings.

0 commit comments

Comments
 (0)