Add `is_empty()` method to `QueueClient`

[adendek]

Is your feature request related to a problem? Please describe.

I am frustrated by the lack of a straightforward, reliable way to check if a queue is truly empty. Currently, a common workaround is to attempt to receive a message with a short visibility timeout:

def is_empty(self) -> bool: 
   return self.queue_client.receive_message(visibility_timeout=0) is None

However, this approach is flawed. It yields a false positive if the queue contains messages that are currently invisible (e.g., being processed by another worker). In this scenario, the method returns True, even though the queue is not technically empty and will have messages available again once visibility timeouts expire.

Describe the solution you'd like

I would like a built-in is_empty() method (or a similar property) on the QueueClient class. This method should ideally check the metadata of the queue to determine if there are any messages present, regardless of their visibility state.

Additional context

In a multi-consumer environment, the current "receive" hack causes logic errors where a system might shut down or trigger a "cleanup" task believing the work is done, while other consumers are still holding onto messages.

Below is the current wrapper implementation I am using, which highlights the need for a more robust built-in solution:

class MyQueueClient():
   def __init__(self, queue_name: str = "my-queue-test", connection_option: str = ""):
    self.queue_service_client = (
        self.__get_queue_service_from_connection_string(connection_option)
        if self.__authenticate_fron_connection_str(connection_option)
        else self.__get_queue_service_from_default_credential(connection_option)
    )

    try:
        self.queue_client = self.queue_service_client.create_queue(queue_name)
    except ResourceExistsError:
        self.queue_client = self.queue_service_client.get_queue_client(queue_name)

# This is the problematic implementation I am trying to replace:
def is_empty(self) -> bool: 
   return self.queue_client.receive_message(visibility_timeout=0) is None

[xiangyan99]

Thanks for reaching out.

Could you clarify what azure library do you use? and which version?

[github-actions]

Hi @adendek. Thank you for opening this issue and giving us the opportunity to assist. To help our team better understand your issue and the details of your scenario please provide a response to the question asked above or the information requested above. This will help us more accurately address your issue.

[adendek]

Hello @xiangyan99,

My apologies for missing those details in the initial post. While this is a general feature request for the SDK's abstraction layer, here are the environment details for my current implementation:

• Python: 3.10.10
• azure-storage-queue: 12.12.0
• azure-identity: 1.16.1

---

From the library perspective I am using the following methods (producer side):

# create queue service client
queue_service = QueueServiceClient.from_connection_string(connection_str)

# create queue client
try:
     logger.info(f"Creating Az Queue: {queue_name} if does not exist.")
     queue_client = self.queue_service_client.create_queue(queue_name)
     logger.info(f"Queue: {queue_name} created.")
except ResourceExistsError:
      logger.info(f"Queue: {queue_name} already exist.")
      queue_client = self.queue_service_client.get_queue_client(queue_name)

# send message
message = queue_client.send_message(message_content)

In the meantime, the consumer looks like:

# build a queue client similarly to the producer
message = queue_client.receive_message(visibility_timeout=visibility_timeout)
do_sth_with_message(message)
queue_client.delete_message(msg, pop_receipt=msg.pop_receipt)

Now inside the producer we would like to check whether the queue is empty, ideally the SDK should provide the following method:

if queue_client.is_empty():
   do_sth_else()
else:  
   wait_for_all_statuses()

However the method is_empty is not supported.