# SkillPatch skill: azure-storage-queue-py

This skill provides a comprehensive guide for using the Azure Queue Storage SDK for Python. It covers authentication with DefaultAzureCredential, lifecycle management, queue operations (create, list, delete), and message workflows including sending, receiving, peeking, and deleting messages. It is designed to help agents implement reliable asynchronous message queuing and task distribution on Azure.

You (the agent) were given this URL and asked to use this skill. This is a **public** skill — no SkillPatch account, API key, or prior setup is required. Two ways to use it:

**1. Use it right now** — the skill's complete file tree (SKILL.md + all reference files) is inlined below; read `SKILL.md` first, then follow it, consulting the other files as it directs.

**2. Install the exact package onto disk** (recommended if you can run a shell — this reproduces the skill byte-for-byte, including any binary assets that can't be inlined):

```bash
mkdir -p .claude/skills/azure-storage-queue-py
curl -sSL https://skillpatch.dev/install_skill/azure-storage-queue-py | tar -xz -C .claude/skills/
```

(`.claude/skills/` is Claude Code's convention; use whatever directory your agent loads skills from.)


---

## Skill files (1)

- `SKILL.md`


### `SKILL.md`

````markdown
---
name: azure-storage-queue-py
description: |
  Azure Queue Storage SDK for Python. Use for reliable message queuing, task distribution, and asynchronous processing.
  Triggers: "queue storage", "QueueServiceClient", "QueueClient", "message queue", "dequeue".
license: MIT
metadata:
  author: Microsoft
  version: "1.0.0"
  package: azure-storage-queue
---

# Azure Queue Storage SDK for Python

Simple, cost-effective message queuing for asynchronous communication.

## Installation

```bash
pip install azure-storage-queue azure-identity
```

## Environment Variables

```bash
AZURE_STORAGE_ACCOUNT_URL=https://<account>.queue.core.windows.net  # Required for all auth methods
AZURE_TOKEN_CREDENTIALS=prod # Required only if DefaultAzureCredential is used in production
```

## Authentication & Lifecycle

> **🔑 Two rules apply to every code sample below:**
>
> 1. **Prefer `DefaultAzureCredential`.** It works locally (Azure CLI / VS Code / Developer CLI) and in Azure (managed identity, workload identity) with no code change. Avoid connection strings, account/API keys — they bypass Entra audit and rotation.
>    - Local dev: `DefaultAzureCredential` works as-is.
>    - Production: set `AZURE_TOKEN_CREDENTIALS=prod` (or `AZURE_TOKEN_CREDENTIALS=<specific_credential>`) to constrain the credential chain to production-safe credentials.
> 2. **Wrap every client in a context manager** so HTTP transports, sockets, and token caches are released deterministically:
>    - Sync: `with <Client>(...) as client:`
>    - Async: `async with <Client>(...) as client:` **and** `async with DefaultAzureCredential() as credential:` (from `azure.identity.aio`)
>
> Snippets may abbreviate this setup, but production code should always follow both rules.

```python
from azure.identity import DefaultAzureCredential, ManagedIdentityCredential
from azure.storage.queue import QueueServiceClient, QueueClient

# Local dev: DefaultAzureCredential. Production: set AZURE_TOKEN_CREDENTIALS=prod or AZURE_TOKEN_CREDENTIALS=<specific_credential>
credential = DefaultAzureCredential(require_envvar=True)
# Or use a specific credential directly in production:
# See https://learn.microsoft.com/python/api/overview/azure/identity-readme?view=azure-python#credential-classes
# credential = ManagedIdentityCredential()
account_url = "https://<account>.queue.core.windows.net"

# Service client
with QueueServiceClient(account_url=account_url, credential=credential) as service_client:
    # Use service_client here (see following sections for operations)
    ...

# Queue client
with QueueClient(account_url=account_url, queue_name="myqueue", credential=credential) as queue_client:
    # Use queue_client here (see following sections for operations)
    ...
```

## Queue Operations

```python
# Create queue
service_client.create_queue("myqueue")

# Get queue client
queue_client = service_client.get_queue_client("myqueue")

# Delete queue
service_client.delete_queue("myqueue")

# List queues
for queue in service_client.list_queues():
    print(queue.name)
```

## Send Messages

```python
# Send message (string)
queue_client.send_message("Hello, Queue!")

# Send with options
queue_client.send_message(
    content="Delayed message",
    visibility_timeout=60,  # Hidden for 60 seconds
    time_to_live=3600       # Expires in 1 hour
)

# Send JSON
import json
data = {"task": "process", "id": 123}
queue_client.send_message(json.dumps(data))
```

## Receive Messages

```python
# Receive messages (makes them invisible temporarily)
messages = queue_client.receive_messages(
    messages_per_page=10,
    visibility_timeout=30  # 30 seconds to process
)

for message in messages:
    print(f"ID: {message.id}")
    print(f"Content: {message.content}")
    print(f"Dequeue count: {message.dequeue_count}")
    
    # Process message...
    
    # Delete after processing
    queue_client.delete_message(message)
```

## Peek Messages

```python
# Peek without hiding (doesn't affect visibility)
messages = queue_client.peek_messages(max_messages=5)

for message in messages:
    print(message.content)
```

## Update Message

```python
# Extend visibility or update content
messages = queue_client.receive_messages()
for message in messages:
    # Extend timeout (need more time)
    queue_client.update_message(
        message,
        visibility_timeout=60
    )
    
    # Update content and timeout
    queue_client.update_message(
        message,
        content="Updated content",
        visibility_timeout=60
    )
```

## Delete Message

```python
# Delete after successful processing
messages = queue_client.receive_messages()
for message in messages:
    try:
        # Process...
        queue_client.delete_message(message)
    except Exception:
        # Message becomes visible again after timeout
        pass
```

## Clear Queue

```python
# Delete all messages
queue_client.clear_messages()
```

## Queue Properties

```python
# Get queue properties
properties = queue_client.get_queue_properties()
print(f"Approximate message count: {properties.approximate_message_count}")

# Set/get metadata
queue_client.set_queue_metadata(metadata={"environment": "production"})
properties = queue_client.get_queue_properties()
print(properties.metadata)
```

## Async Client

```python
from azure.storage.queue.aio import QueueServiceClient, QueueClient
from azure.identity.aio import DefaultAzureCredential

async def queue_operations():
    credential = DefaultAzureCredential()
    
    async with QueueClient(
        account_url="https://<account>.queue.core.windows.net",
        queue_name="myqueue",
        credential=credential
    ) as client:
        # Send
        await client.send_message("Async message")
        
        # Receive
        async for message in client.receive_messages():
            print(message.content)
            await client.delete_message(message)

import asyncio
asyncio.run(queue_operations())
```

## Base64 Encoding

```python
from azure.storage.queue import QueueClient, BinaryBase64EncodePolicy, BinaryBase64DecodePolicy

# For binary data
with QueueClient(
    account_url=account_url,
    queue_name="myqueue",
    credential=credential,
    message_encode_policy=BinaryBase64EncodePolicy(),
    message_decode_policy=BinaryBase64DecodePolicy()
) as queue_client:
    # Send bytes
    queue_client.send_message(b"Binary content")
```

## Best Practices

1. **Pick sync OR async and stay consistent.** Do not mix `azure.xxx` sync clients with `azure.xxx.aio` async clients in the same call path. Choose one mode per module.
2. **Always use context managers for clients and async credentials.** Wrap every client in `with Client(...) as client:` (sync) or `async with Client(...) as client:` (async). For async `DefaultAzureCredential` from `azure.identity.aio`, also use `async with credential:` so tokens and transports are cleaned up.
3. **Use `DefaultAzureCredential`** for portable auth across local dev and Azure (avoid connection strings / API keys when possible).
4. **Delete messages after processing** to prevent reprocessing
5. **Set appropriate visibility timeout** based on processing time
6. **Handle `dequeue_count`** for poison message detection
7. **Use async client** for high-throughput scenarios
8. **Use `peek_messages`** for monitoring without affecting queue
9. **Set `time_to_live`** to prevent stale messages
10. **Consider Service Bus** for advanced features (sessions, topics)

````