tessl/pypi-azure-storage-queue
Microsoft Azure Queue Storage Client Library for Python providing comprehensive message queuing capabilities for distributed applications.
—
Pending
Quality
Pending
Does it follow best practices?
Impact
Pending
No eval scenarios have been run
queue-operations.mddocs/
Queue Operations
Queue-specific operations for managing individual queues, their properties, metadata, and access policies. The QueueClient provides comprehensive queue-level management capabilities.
Capabilities
QueueClient Creation
Create queue clients using various authentication methods and initialization patterns.
class QueueClient:
def __init__(
self,
account_url: str,
queue_name: str,
credential=None,
*,
api_version: Optional[str] = None,
secondary_hostname: Optional[str] = None,
message_encode_policy=None,
message_decode_policy=None,
audience: Optional[str] = None,
**kwargs
):
"""
Create a QueueClient for a specific queue.
Parameters:
- account_url: Queue service endpoint URL
- queue_name: Name of the target queue
- credential: Authentication credential
- api_version: Storage service API version
- secondary_hostname: Secondary hostname for read access
- message_encode_policy: Custom message encoding policy
- message_decode_policy: Custom message decoding policy
- audience: Token audience for Azure AD authentication
"""
@classmethod
def from_queue_url(
cls,
queue_url: str,
credential=None,
**kwargs
) -> 'QueueClient':
"""
Create client from complete queue URL.
Parameters:
- queue_url: Complete URL to the queue
- credential: Authentication credential
Returns:
QueueClient instance
"""
@classmethod
def from_connection_string(
cls,
conn_str: str,
queue_name: str,
credential=None,
**kwargs
) -> 'QueueClient':
"""
Create client from connection string and queue name.
Parameters:
- conn_str: Storage account connection string
- queue_name: Target queue name
- credential: Optional additional credential
Returns:
QueueClient instance
"""Queue Lifecycle Management
Create, delete, and check queue existence.
def create_queue(
self,
*,
metadata: Optional[Dict[str, str]] = None,
timeout: Optional[int] = None,
**kwargs
) -> None:
"""
Create the queue if it does not exist.
Parameters:
- metadata: User-defined metadata key-value pairs
- timeout: Request timeout in seconds
Raises:
ResourceExistsError: If queue already exists (when checking preconditions)
"""
def delete_queue(
self,
*,
timeout: Optional[int] = None,
**kwargs
) -> None:
"""
Delete the queue and all its messages.
Parameters:
- timeout: Request timeout in seconds
Raises:
ResourceNotFoundError: If queue does not exist
"""Queue Properties and Metadata
Get and set queue properties and user-defined metadata.
def get_queue_properties(
self,
*,
timeout: Optional[int] = None,
**kwargs
) -> QueueProperties:
"""
Get queue properties including metadata and message count.
Parameters:
- timeout: Request timeout in seconds
Returns:
QueueProperties with name, metadata, and approximate message count
Raises:
ResourceNotFoundError: If queue does not exist
"""
def set_queue_metadata(
self,
metadata: Optional[Dict[str, str]] = None,
*,
timeout: Optional[int] = None,
**kwargs
) -> Dict[str, Any]:
"""
Set user-defined metadata for the queue.
Parameters:
- metadata: Metadata key-value pairs (keys must be valid HTTP header names)
- timeout: Request timeout in seconds
Returns:
Dictionary with response headers including ETag and last modified time
Raises:
ResourceNotFoundError: If queue does not exist
"""
# Property for queue name
queue_name: str
"""The name of the queue."""Access Policy Management
Manage stored access policies for fine-grained permissions control.
def get_queue_access_policy(
self,
*,
timeout: Optional[int] = None,
**kwargs
) -> Dict[str, AccessPolicy]:
"""
Get stored access policies for the queue.
Parameters:
- timeout: Request timeout in seconds
Returns:
Dictionary mapping policy IDs to AccessPolicy objects
Raises:
ResourceNotFoundError: If queue does not exist
"""
def set_queue_access_policy(
self,
signed_identifiers: Dict[str, AccessPolicy],
*,
timeout: Optional[int] = None,
**kwargs
) -> None:
"""
Set stored access policies for the queue.
Parameters:
- signed_identifiers: Dictionary of policy ID to AccessPolicy mappings (max 5 policies)
- timeout: Request timeout in seconds
Raises:
ResourceNotFoundError: If queue does not exist
"""Usage Examples
Basic Queue Operations
from azure.storage.queue import QueueClient
# Create queue client
queue_client = QueueClient.from_connection_string(
conn_str="your_connection_string",
queue_name="myqueue"
)
# Create queue with metadata
queue_client.create_queue(metadata={"purpose": "test-queue", "env": "development"})
# Get queue properties
properties = queue_client.get_queue_properties()
print(f"Queue: {properties.name}")
print(f"Messages: {properties.approximate_message_count}")
print(f"Metadata: {properties.metadata}")
# Update metadata
queue_client.set_queue_metadata({"purpose": "production-queue", "env": "prod"})
# Delete queue
queue_client.delete_queue()Access Policy Management
from azure.storage.queue import AccessPolicy, QueueSasPermissions
from datetime import datetime, timedelta
# Create access policies
read_policy = AccessPolicy(
permission=QueueSasPermissions(read=True, process=True),
expiry=datetime.utcnow() + timedelta(hours=24),
start=datetime.utcnow()
)
write_policy = AccessPolicy(
permission=QueueSasPermissions(add=True, update=True),
expiry=datetime.utcnow() + timedelta(hours=12),
start=datetime.utcnow()
)
# Set access policies
queue_client.set_queue_access_policy({
"read-access": read_policy,
"write-access": write_policy
})
# Get access policies
policies = queue_client.get_queue_access_policy()
for policy_id, policy in policies.items():
print(f"Policy: {policy_id}, Permissions: {policy.permission}")Error Handling
from azure.storage.queue import QueueClient
from azure.core.exceptions import ResourceExistsError, ResourceNotFoundError
queue_client = QueueClient.from_connection_string(conn_str, "myqueue")
try:
queue_client.create_queue()
print("Queue created successfully")
except ResourceExistsError:
print("Queue already exists")
try:
properties = queue_client.get_queue_properties()
print(f"Queue has {properties.approximate_message_count} messages")
except ResourceNotFoundError:
print("Queue does not exist")Types
Queue Property Types
class QueueProperties:
name: str
metadata: Optional[Dict[str, str]]
approximate_message_count: Optional[int]
@classmethod
def _from_generated(cls, generated) -> 'QueueProperties': ...
class AccessPolicy:
permission: Optional[Union[QueueSasPermissions, str]]
expiry: Optional[Union[datetime, str]]
start: Optional[Union[datetime, str]]
class QueueSasPermissions:
read: bool # Read metadata and properties, peek messages
add: bool # Add messages to queue
update: bool # Update messages in queue
process: bool # Get and delete messages
def __str__(self) -> str: ...
@classmethod
def from_string(cls, permission: str) -> 'QueueSasPermissions': ...Install with Tessl CLI
npx tessl i tessl/pypi-azure-storage-queue