How to work with Azure Service Bus without context manager (Python)

[EvgeniiTitov]

Hi!

I find it super confusing to work with Azure service bus queues. As a result, I am afraid I am making mistakes without even understanding this and get unexpected behavior.
I want to have a class, say, Consumer where in the constructor I instantiate all the necessary things such as ServiceBusClient, AutoLockRenewer, and Receiver. Then, say, I have 2 methods - get_message() and acknowledge_message(). My other code receives a message, attempts to process it, and then if successfully processed acknowledges the message.

ALL examples use context managers, which is not the way we'd like to go. My attempts to do it my way resulted in the following monstrosity, which is obviously wrong:

class CustomServiceBusClient(ServiceBusClient):
    def enter(self):
        if self._connection_sharing:
            self._create_uamqp_connection()

    def exit(self):
        self.close()


class AzureConsumer:
    def __init__(
            self,
            connection_str: t.Optional[str] = None,
            queue_name: t.Optional[str] = None,
            timeout: int = 10,
            logging_enable: bool = False,
    ) -> None:
        self._conn_str = connection_str
        self._queue_name = queue_name
        self._timeout = timeout
        self._logging_enable = logging_enable
        self._being_processed_messages = {}
        try:
            self._client = CustomServiceBusClient.from_connection_string(
                conn_str=self._conn_str, logging_enable=self._logging_enable
            )
        except Exception as e:
            raise e
        else:
            self._client.enter()
        self._lock_renewer = AutoLockRenewer(max_workers=2)
        self._receiver = self._client.get_queue_receiver(
            self._queue_name, max_wait_time=self._timeout
        )

    def get_message(self) -> t.Tuple[t.Optional[str], t.Optional[str]]:
        try:
            msg = next(iter(self._receiver))
        except StopIteration:
            return None, None
        # Register the message with the lock renewer
        self._lock_renewer.register(
            self._receiver, msg, max_lock_renewal_duration=6000.0
        )
        message_id = str(msg.message_id)
        message_content = str(msg.message)
        self._being_processed_messages[message_id] = msg
        return message_content, message_id

    def acknowledge_message(self, message_id: str) -> None:
        if message_id not in self._being_processed_messages:
            raise KeyError(
                f"The key {message_id} doesn't belong to any messages!"
            )
        self._receiver.complete_message(
            self._being_processed_messages.pop(message_id)
        )

OR something among those lines (the example is incorrect, I got completely lost)

class Consumer:
    QUEUE_NAME = ""
    CONNECTION_STR = ""

    def __init__(self):
        self._bus_client: ServiceBusClient = ServiceBusClient.from_connection_string(
            conn_str=Consumer.CONNECTION_STR
        )
        self._lock_renewer = AutoLockRenewer(
            max_lock_renewal_duration=300,
            max_workers=4,
        )

    def get_message(self) -> t.Generator:
        with self._bus_client.get_queue_receiver(Consumer.QUEUE_NAME) as rec:
            for message_obj in rec:
                self._lock_renewer.register(
                    rec, message_obj, max_lock_renewal_duration=300
                )
                yield message_obj.message
                ok: bool = yield
                if ok:
                    rec.complete_message(message_obj)

Could anyone explain how to implement something similar or direct me to the appropriate resource, please? I've tried my best to google, but there are so few examples out there and they all seem to use the context manager approach.

Thank you, much appreciated.

ET

P.S.
Apologies if thats the wrong place to ask it.

[rakshith91]

Thanks for the question!! We'll look into this asap

[yunhaoling]

hey @EvgeniiTitov , you could use ServiceBusClient.close method to shutdown the client or use ServiceBusRecever.close method to shutdown the spawned receiver.

sb_client = ServiceBusClient.from_connection_string(...)
receiver = sb_client.get_queue_receiver(...)

# your receiving code

receiver.close()
sb_client.close()

Let me know if you need more help!

[ghost]

Hi @EvgeniiTitov. Thank you for opening this issue and giving us the opportunity to assist. We believe that this has been addressed. If you feel that further discussion is needed, please add a comment with the text “/unresolve” to remove the “issue-addressed” label and continue the conversation.