Google pubsub SubscriberClient AttributeError: объект «генератор» не имеет атрибута «add_done_callback» ⇐ Python

[Anonymous]

Я работаю над проектом и должен использовать некоторые сообщения из моей подписки GCP Pubsub. Локально я могу нормально получать сообщения.
Но когда я выпускаю изменения в контейнер в облаке (что-то вроде промежуточного окружения), я получаю эту ошибку:

Код: Выделить всё

2024-05-14 15:50:15,379  ERROR [google.api_core.bidi:_thread_main:678] [pid=1] [tname=Thread-ConsumeBidirectionalStream] [cluster=mt2] [1.65.1] Thread-ConsumeBidirectionalStream caught unexpected exception 'generator' object has no attribute 'add_done_callback' and will exit.
Traceback (most recent call last):
File "/opt/project/lib/python3.9/site-packages/google/api_core/bidi.py", line 644, in _thread_main
self._bidi_rpc.open()
File "/opt/project/lib/python3.9/site-packages/google/api_core/bidi.py", line 294, in open
call._wrapped.add_done_callback(self._on_call_done)
AttributeError: 'generator' object has no attribute 'add_done_callback'

Мой код прост: контейнер докера с точкой входа в bash, запускающий файл Python, который инициализирует потребителя.
init.py

Код: Выделить всё

def main():
"""
Entrypoint for all consumers
"""

args = get_args()
config_settings = get_config_section(args.config_filename)
consumer = BaseGooglePubSubConsumer(config_settings)
consumer.run()

if __name__ == "__main__":
main()

consumer.py

Код: Выделить всё

from functools import cached_property

from google.cloud.pubsub_v1 import SubscriberClient

class BaseGooglePubSubConsumer(BaseConsumer):

@cached_property
def consumer(self) -> SubscriberClient:
json_account_info = self._get_service_account_info()
credentials = service_account.Credentials.from_service_account_info(json_account_info)
return SubscriberClient(credentials=credentials)

def run(self):
"""
Consumes messages from a Pub/Sub Subscriber.
"""
streaming_pull_future = self.consumer.subscribe(self._get_subscription_path(), callback=self.process_message)
with self.consumer:
try:
streaming_pull_future.result()
except Exception as exc:
le.errors = "Failed to process message"
logger.error(le, exc_info=exc)
streaming_pull_future.cancel()
streaming_pull_future.result()
finally:
le.end()
def process_message(self, message: Message) -> None:
# message.ack()
logger.info(message)

Я прочитал об ошибке и обнаружил аналогичную проблему без ответа: https://github.com/open-telemetry/opent ... ssues/1180
п>

Подробнее здесь: https://stackoverflow.com/questions/784 ... no-attribu