CtrlK
BlogDocsLog inGet started
Tessl Logo

tessl/pypi-azure-storage-blob

Microsoft Azure Blob Storage Client Library for Python providing comprehensive APIs for blob storage operations.

Pending

Quality

Pending

Does it follow best practices?

Impact

Pending

No eval scenarios have been run

Overview
Eval results
Files

container-client.mddocs/

Container Management

Container-level operations including creation, deletion, property management, access control, and blob operations within a container. The ContainerClient provides comprehensive management capabilities for Azure Blob Storage containers.

Capabilities

Client Creation

Create ContainerClient instances with various authentication methods and from different sources including URLs and connection strings.

class ContainerClient:
    def __init__(self, account_url: str, container_name: str, credential=None, **kwargs):
        """
        Create a ContainerClient.
        
        Args:
            account_url (str): The URL to the storage account
            container_name (str): Name of the container
            credential: The credentials for authentication. Can be:
                - str: Account key or SAS token string
                - dict: Account name and key mapping
                - AzureNamedKeyCredential: Named key credential  
                - AzureSasCredential: SAS credential
                - TokenCredential: Azure AD token credential
                - None: For anonymous access or SAS URLs
        """

    @classmethod
    def from_container_url(cls, container_url: str, credential=None, **kwargs) -> 'ContainerClient':
        """
        Create a ContainerClient from a container URL.
        
        Args:
            container_url (str): Complete URL to the container
            credential: Optional credential for authentication
            
        Returns:
            ContainerClient: Configured client instance
        """

    @classmethod  
    def from_connection_string(cls, conn_str: str, container_name: str, credential=None, **kwargs) -> 'ContainerClient':
        """
        Create a ContainerClient from a connection string.
        
        Args:
            conn_str (str): Azure Storage connection string
            container_name (str): Name of the container
            credential: Optional credential to override connection string auth
            
        Returns:
            ContainerClient: Configured client instance
        """

Container Lifecycle

Create, delete, and check container existence. These operations manage the container lifecycle within Azure Blob Storage.

def create_container(self, metadata=None, public_access=None, **kwargs) -> dict:
    """
    Create the container.
    
    Args:
        metadata (dict, optional): Container metadata as key-value pairs
        public_access (PublicAccess, optional): Public access level for the container
        
    Returns:
        dict: Container creation response with ETag and last modified time
    """

def delete_container(self, **kwargs) -> None:
    """
    Delete the container and all its blobs.
    
    Keyword Args:
        lease (BlobLeaseClient or str, optional): Required if container has an active lease
        if_modified_since (datetime, optional): Delete only if modified since this time
        if_unmodified_since (datetime, optional): Delete only if not modified since this time
        etag (str, optional): Delete only if ETag matches
        match_condition (MatchConditions, optional): ETag matching condition
    """

def exists(self, **kwargs) -> bool:
    """
    Check whether the container exists.
    
    Returns:
        bool: True if container exists, False otherwise
    """

Container Properties and Metadata

Retrieve and modify container properties, metadata, and access policies.

def get_container_properties(self, **kwargs) -> ContainerProperties:
    """
    Get container properties and metadata.
    
    Returns:
        ContainerProperties: Container properties including metadata, lease state, and access policies
    """

def set_container_metadata(self, metadata=None, **kwargs) -> dict:
    """
    Set container metadata.
    
    Args:
        metadata (dict, optional): Container metadata as key-value pairs
        
    Returns:
        dict: Response with ETag and last modified time
    """

def get_container_access_policy(self, **kwargs) -> dict:
    """
    Get the container's stored access policies.
    
    Returns:
        dict: Dictionary containing:
            - public_access: PublicAccess level
            - signed_identifiers: List of AccessPolicy objects
    """

def set_container_access_policy(self, signed_identifiers=None, public_access=None, **kwargs) -> dict:
    """
    Set the container's access policy and public access level.
    
    Args:
        signed_identifiers (dict, optional): Stored access policies by identifier
        public_access (PublicAccess, optional): Public access level
        
    Returns:
        dict: Response with ETag and last modified time
    """

Container Leasing

Acquire and manage leases on containers to prevent concurrent modifications.

def acquire_lease(self, lease_duration=-1, lease_id=None, **kwargs) -> BlobLeaseClient:
    """
    Acquire a lease on the container.
    
    Args:
        lease_duration (int): Lease duration in seconds (-1 for infinite)
        lease_id (str, optional): Proposed lease identifier
        
    Returns:
        BlobLeaseClient: Lease client for managing the lease
    """

Blob Listing and Traversal

List blobs within the container with various filtering and inclusion options, including hierarchical directory-style traversal.

def list_blobs(self, name_starts_with=None, include=None, **kwargs) -> ItemPaged[BlobProperties]:
    """
    List blobs in the container.
    
    Args:
        name_starts_with (str, optional): Filter blobs by name prefix
        include (list, optional): Additional properties to include:
            - 'snapshots': Include blob snapshots
            - 'metadata': Include blob metadata  
            - 'uncommittedblobs': Include uncommitted blobs
            - 'copy': Include copy properties
            - 'deleted': Include soft-deleted blobs
            - 'tags': Include blob index tags
            - 'versions': Include blob versions
            - 'immutabilitypolicy': Include immutability policy
            - 'legalhold': Include legal hold status
            - 'deletedwithversions': Include deleted blobs with versions
            
    Returns:
        ItemPaged[BlobProperties]: Paginated list of blob properties
    """

def list_blob_names(self, **kwargs) -> ItemPaged[str]:
    """  
    List blob names only (more efficient than full properties).
    
    Returns:
        ItemPaged[str]: Paginated list of blob names
    """

def walk_blobs(self, name_starts_with=None, include=None, delimiter='/', **kwargs) -> ItemPaged[Union[BlobProperties, BlobPrefix]]:
    """
    Walk blobs in a directory-like hierarchy using delimiter.
    
    Args:
        name_starts_with (str, optional): Filter by name prefix
        include (list, optional): Additional properties to include
        delimiter (str): Delimiter for hierarchical listing (default '/')
        
    Returns:
        ItemPaged[Union[BlobProperties, BlobPrefix]]: Paginated list of blobs and prefixes
    """

def find_blobs_by_tags(self, filter_expression: str, **kwargs) -> ItemPaged['FilteredBlob']:
    """
    Find blobs in the container using tag-based filtering.
    
    Args:
        filter_expression (str): OData filter expression for blob tags
        
    Returns:
        ItemPaged[FilteredBlob]: Paginated list of matching blobs
    """

Blob Operations

Upload, download, and delete blobs through the container client for convenient blob management.

def upload_blob(self, name: str, data, blob_type='BlockBlob', **kwargs) -> BlobClient:
    """
    Upload data as a blob in the container.
    
    Args:
        name (str): Name for the blob
        data: Data to upload (bytes, str, or file-like object)
        blob_type (BlobType): Type of blob to create
        
    Keyword Args:
        length (int, optional): Number of bytes to upload
        metadata (dict, optional): Blob metadata
        content_type (str, optional): MIME content type
        content_encoding (str, optional): Content encoding  
        content_language (str, optional): Content language
        content_disposition (str, optional): Content disposition
        cache_control (str, optional): Cache control header
        content_md5 (bytes, optional): MD5 hash of content
        validate_content (bool): Validate content integrity
        max_concurrency (int): Maximum concurrent uploads
        overwrite (bool): Whether to overwrite existing blob
        standard_blob_tier (StandardBlobTier, optional): Access tier
        tags (dict, optional): Blob index tags
        
    Returns:
        BlobClient: Client for the uploaded blob
    """

def download_blob(self, blob: str, offset=None, length=None, **kwargs) -> StorageStreamDownloader:
    """
    Download a blob from the container.
    
    Args:
        blob (str): Name of the blob to download
        offset (int, optional): Start byte position
        length (int, optional): Number of bytes to download
        
    Returns:
        StorageStreamDownloader: Streaming downloader for blob content
    """

def delete_blob(self, blob: str, delete_snapshots=None, **kwargs) -> None:
    """
    Delete a blob from the container.
    
    Args:
        blob (str): Name of the blob to delete
        delete_snapshots (str, optional): How to handle snapshots:
            - 'include': Delete blob and all snapshots
            - 'only': Delete snapshots but not the blob
    """

Batch Operations

Perform efficient batch operations on multiple blobs within the container.

def delete_blobs(self, *blobs, **kwargs) -> Iterator[HttpResponse]:
    """
    Delete multiple blobs in a batch operation.
    
    Args:
        *blobs: Blob names or BlobProperties objects to delete
        
    Keyword Args:
        delete_snapshots (str, optional): How to handle snapshots
        raise_on_any_failure (bool): Whether to raise on any failure
        
    Returns:
        Iterator[HttpResponse]: Response for each delete operation
    """

def set_standard_blob_tier_blobs(self, *blobs, **kwargs) -> Iterator[HttpResponse]:
    """
    Set access tier for multiple standard blobs in batch.
    
    Args:
        *blobs: Tuples of (blob_name, standard_blob_tier) or BlobProperties with tier
        
    Returns:
        Iterator[HttpResponse]: Response for each tier operation
    """

def set_premium_page_blob_tier_blobs(self, *blobs, **kwargs) -> Iterator[HttpResponse]:
    """
    Set access tier for multiple premium page blobs in batch.
    
    Args:
        *blobs: Tuples of (blob_name, premium_page_blob_tier) or BlobProperties with tier
        
    Returns:
        Iterator[HttpResponse]: Response for each tier operation
    """

Blob Client Factory

Create blob clients for individual blob operations within the container.

def get_blob_client(self, blob: str, snapshot=None) -> 'BlobClient':
    """
    Get a BlobClient for a specific blob in this container.
    
    Args:
        blob (str): Name of the blob
        snapshot (str, optional): Blob snapshot identifier
        
    Returns:
        BlobClient: Client for blob operations
    """

Account Information

Retrieve storage account information through the container client.

def get_account_information(self, **kwargs) -> dict:
    """
    Get information about the storage account.
    
    Returns:
        dict: Account information including account kind and SKU
    """

Supporting Data Types

class PublicAccess:
    """Container public access levels."""
    OFF: str        # No public access
    BLOB: str       # Public read access for blobs only  
    CONTAINER: str  # Public read access for container and blobs

class ContainerProperties:
    """Container properties and metadata."""
    name: str
    last_modified: datetime
    etag: str
    lease: LeaseProperties
    public_access: PublicAccess
    has_immutability_policy: bool
    deleted: bool
    version: str
    has_legal_hold: bool
    metadata: dict
    encryption_scope: ContainerEncryptionScope
    immutable_storage_with_versioning_enabled: bool

class BlobPrefix:
    """Blob name prefix for hierarchical listing."""
    name: str
    service_endpoint: str
    container_name: str

class AccessPolicy:
    """Stored access policy for container."""
    permission: str
    expiry: datetime
    start: datetime

class LeaseProperties:
    """Container or blob lease information."""
    status: str
    state: str
    duration: str
    
class ContainerEncryptionScope:
    """Container encryption scope configuration."""
    default_encryption_scope: str
    deny_encryption_scope_override: bool

Install with Tessl CLI

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

docs

async-operations.md

blob-client.md

blob-types-tiers.md

container-client.md

index.md

sas-generation.md

service-client.md

utility-functions.md

tile.json