CtrlK
BlogDocsLog inGet started
Tessl Logo

tessl/pypi-azure-mgmt-storage

Microsoft Azure Storage Management Client Library for Python

—

Pending

Quality

Pending

Does it follow best practices?

Impact

Pending

No eval scenarios have been run

Overview
Eval results
Files

queue-storage.mddocs/

Queue Storage Management

Azure Queue service configuration and individual queue management operations. Azure Queue storage provides reliable messaging between application components with support for large numbers of messages.

Capabilities

Queue Service Configuration

Configure queue service properties including CORS rules, logging, and metrics settings.

class QueueServicesOperations:
    def get_service_properties(
        self,
        resource_group_name: str,
        account_name: str
    ) -> QueueServiceProperties:
        """
        Gets the properties of a storage account's Queue service.
        
        Parameters:
        - resource_group_name: Name of the resource group
        - account_name: Name of the storage account
        
        Returns:
        QueueServiceProperties with current configuration
        """
    
    def set_service_properties(
        self,
        resource_group_name: str,
        account_name: str,
        parameters: QueueServiceProperties
    ) -> QueueServiceProperties:
        """
        Sets the properties of a storage account's Queue service.
        
        Parameters:
        - resource_group_name: Name of the resource group
        - account_name: Name of the storage account
        - parameters: Queue service properties to configure
        
        Returns:
        Updated QueueServiceProperties
        """
    
    def list(
        self,
        resource_group_name: str,
        account_name: str
    ) -> ListQueueServices:
        """
        Lists all queue services for a storage account.
        
        Parameters:
        - resource_group_name: Name of the resource group
        - account_name: Name of the storage account
        
        Returns:
        ListQueueServices containing queue services
        """

Usage example:

from azure.mgmt.storage.models import (
    QueueServiceProperties, CorsRules, CorsRule
)

# Configure queue service with CORS rules
cors_rule = CorsRule(
    allowed_origins=["https://myapp.com", "https://*.myapp.com"],
    allowed_methods=["GET", "POST", "PUT", "DELETE"],
    allowed_headers=["*"],
    exposed_headers=["x-ms-request-id", "x-ms-version"],
    max_age_in_seconds=3600
)

queue_properties = QueueServiceProperties(
    properties=QueueServicePropertiesProperties(
        cors=CorsRules(cors_rules=[cors_rule])
    )
)

updated_service = client.queue_services.set_service_properties(
    resource_group_name="my-resource-group",
    account_name="mystorageaccount123",
    parameters=queue_properties
)

# Get current queue service properties
current_properties = client.queue_services.get_service_properties(
    resource_group_name="my-resource-group",
    account_name="mystorageaccount123"
)

print(f"Queue service configured with {len(current_properties.cors.cors_rules)} CORS rules")

Queue Management

Create, configure, and manage individual queues with metadata and access policies.

class QueueOperations:
    def create(
        self,
        resource_group_name: str,
        account_name: str,
        queue_name: str,
        queue: StorageQueue
    ) -> StorageQueue:
        """
        Creates a new queue with the specified name.
        
        Parameters:
        - resource_group_name: Name of the resource group
        - account_name: Name of the storage account
        - queue_name: Name of the queue (3-63 chars, lowercase letters/numbers/hyphens)
        - queue: Queue properties and metadata
        
        Returns:
        Created StorageQueue
        """
    
    def update(
        self,
        resource_group_name: str,
        account_name: str,
        queue_name: str,
        queue: StorageQueue
    ) -> StorageQueue:
        """
        Updates queue properties or metadata.
        
        Parameters:
        - resource_group_name: Name of the resource group
        - account_name: Name of the storage account
        - queue_name: Name of the queue
        - queue: Updated queue properties
        
        Returns:
        Updated StorageQueue
        """
    
    def get(
        self,
        resource_group_name: str,
        account_name: str,
        queue_name: str
    ) -> StorageQueue:
        """
        Gets the queue with the specified name.
        
        Parameters:
        - resource_group_name: Name of the resource group
        - account_name: Name of the storage account
        - queue_name: Name of the queue
        
        Returns:
        StorageQueue with properties
        """
    
    def delete(
        self,
        resource_group_name: str,
        account_name: str,
        queue_name: str
    ) -> None:
        """
        Deletes the queue with the specified name.
        
        Parameters:
        - resource_group_name: Name of the resource group
        - account_name: Name of the storage account
        - queue_name: Name of the queue to delete
        """
    
    def list(
        self,
        resource_group_name: str,
        account_name: str,
        maxpagesize: Optional[str] = None,
        filter: Optional[str] = None
    ) -> ItemPaged[ListQueueResource]:
        """
        Lists all queues in a storage account.
        
        Parameters:
        - resource_group_name: Name of the resource group
        - account_name: Name of the storage account
        - maxpagesize: Maximum number of results per page
        - filter: OData filter expression
        
        Returns:
        Paginated list of ListQueueResource objects
        """

Usage example:

from azure.mgmt.storage.models import StorageQueue

# Create a queue with metadata
queue_metadata = {
    "purpose": "order-processing",
    "environment": "production",
    "team": "backend"
}

new_queue = StorageQueue(
    properties=StorageQueueProperties(
        metadata=queue_metadata
    )
)

created_queue = client.queue.create(
    resource_group_name="my-resource-group",
    account_name="mystorageaccount123",
    queue_name="order-processing-queue",
    queue=new_queue
)

print(f"Created queue: {created_queue.name}")

# Create multiple queues for different purposes
queue_configs = [
    {"name": "high-priority-tasks", "metadata": {"priority": "high", "retention": "short"}},
    {"name": "low-priority-tasks", "metadata": {"priority": "low", "retention": "long"}},
    {"name": "notification-queue", "metadata": {"type": "notifications", "max-size": "small"}},
    {"name": "audit-log-queue", "metadata": {"type": "audit", "retention": "extended"}}
]

for config in queue_configs:
    queue = StorageQueue(
        properties=StorageQueueProperties(
            metadata=config["metadata"]
        )
    )
    
    client.queue.create(
        resource_group_name="my-resource-group",
        account_name="mystorageaccount123",
        queue_name=config["name"],
        queue=queue
    )

# List all queues
queues = list(client.queue.list(
    resource_group_name="my-resource-group",
    account_name="mystorageaccount123"
))

print(f"Total queues: {len(queues)}")
for queue in queues:
    print(f"Queue: {queue.name}")
    if hasattr(queue, 'metadata') and queue.metadata:
        print(f"  Metadata: {queue.metadata}")

# Update queue metadata
updated_metadata = {
    "purpose": "order-processing",
    "environment": "production", 
    "team": "backend",
    "last-updated": "2024-01-15",
    "max-messages": "1000"
}

updated_queue = StorageQueue(
    properties=StorageQueueProperties(
        metadata=updated_metadata
    )
)

client.queue.update(
    resource_group_name="my-resource-group",
    account_name="mystorageaccount123",
    queue_name="order-processing-queue",
    queue=updated_queue
)

# Get specific queue details
queue_details = client.queue.get(
    resource_group_name="my-resource-group",
    account_name="mystorageaccount123",
    queue_name="order-processing-queue"
)

print(f"Queue details for {queue_details.name}:")
print(f"  ID: {queue_details.id}")
print(f"  Type: {queue_details.type_}")
if hasattr(queue_details, 'properties') and queue_details.properties.metadata:
    print(f"  Metadata: {queue_details.properties.metadata}")

Queue Access Control

Configure access policies and permissions for queue operations.

# Example of using queues with access policies through SAS tokens
from azure.mgmt.storage.models import ServiceSasParameters, Services, SignedResourceTypes

# Generate SAS token for queue access
sas_params = ServiceSasParameters(
    canonical_name=f"/queue/mystorageaccount123/order-processing-queue",
    resource=SignedResource.Q,  # Queue
    permissions=Permissions.RAU,  # Read, Add, Update
    services=Services.Q,  # Queue service
    resource_types=SignedResourceTypes.O,  # Object
    shared_access_expiry_time=datetime.utcnow() + timedelta(hours=24)
)

sas_response = client.storage_accounts.list_service_sas(
    resource_group_name="my-resource-group",
    account_name="mystorageaccount123",
    parameters=sas_params
)

print(f"SAS token for queue access: {sas_response.service_sas_token}")

Types

class QueueServiceProperties:
    """Queue service properties configuration."""
    id: str
    name: str
    type_: str
    properties: QueueServicePropertiesProperties

class QueueServicePropertiesProperties:
    """Properties of queue service."""
    cors: CorsRules

class StorageQueue:
    """Storage queue resource."""
    id: str
    name: str
    type_: str
    properties: StorageQueueProperties

class StorageQueueProperties:
    """Properties of a storage queue."""
    metadata: Dict[str, str]
    approximate_message_count: int

class ListQueueResource:
    """Queue resource in list results."""
    name: str
    properties: StorageQueueProperties

class ListQueueServices:
    """List of queue services."""
    value: List[QueueServiceProperties]

class ListQueue:
    """List of queues."""
    value: List[ListQueueResource]

Install with Tessl CLI

npx tessl i tessl/pypi-azure-mgmt-storage

docs

blob-storage.md

file-storage.md

index.md

policy-management.md

queue-storage.md

security-access.md

storage-accounts.md

storage-tasks.md

table-storage.md

utilities.md

tile.json