Service Client Operations

class BlobServiceClient:
    def __init__(self, account_url: str, credential=None, **kwargs):
        """
        Create a BlobServiceClient.
        
        Args:
            account_url (str): The URL to the blob storage account
            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_connection_string(cls, conn_str: str, credential=None, **kwargs) -> 'BlobServiceClient':
        """
        Create a BlobServiceClient from a connection string.
        
        Args:
            conn_str (str): Azure Storage connection string
            credential: Optional credential to override connection string auth
            
        Returns:
            BlobServiceClient: Configured client instance
        """

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

def get_service_properties(self, **kwargs) -> dict:
    """
    Get service properties for the storage account.
    
    Returns:
        dict: Service properties including logging, metrics, CORS, and retention policies
    """

def set_service_properties(self, analytics_logging=None, hour_metrics=None, minute_metrics=None, cors=None, target_version=None, delete_retention_policy=None, static_website=None, **kwargs) -> None:
    """
    Set service properties for the storage account.
    
    Args:
        analytics_logging (BlobAnalyticsLogging, optional): Logging configuration
        hour_metrics (Metrics, optional): Hour metrics configuration  
        minute_metrics (Metrics, optional): Minute metrics configuration
        cors (list[CorsRule], optional): CORS rules
        target_version (str, optional): Target service version
        delete_retention_policy (RetentionPolicy, optional): Soft delete configuration
        static_website (StaticWebsite, optional): Static website configuration
    """

def get_service_stats(self, **kwargs) -> dict:
    """
    Get statistics for the storage account (geo-replication status).
    
    Returns:
        dict: Service statistics including geo-replication details
    """

def get_user_delegation_key(self, key_start_time, key_expiry_time, **kwargs) -> 'UserDelegationKey':
    """
    Get a user delegation key for creating SAS tokens.
    
    Args:
        key_start_time (datetime): Start time for key validity
        key_expiry_time (datetime): Expiry time for key validity
        
    Returns:
        UserDelegationKey: User delegation key for SAS generation
    """

def list_containers(self, name_starts_with=None, include_metadata=False, **kwargs) -> ItemPaged[ContainerProperties]:
    """
    List containers in the storage account.
    
    Args:
        name_starts_with (str, optional): Filter containers by name prefix
        include_metadata (bool): Whether to include container metadata
        
    Returns:
        ItemPaged[ContainerProperties]: Paginated list of container properties
    """

def create_container(self, name: str, metadata=None, public_access=None, **kwargs) -> ContainerClient:
    """
    Create a new container.
    
    Args:
        name (str): Container name
        metadata (dict, optional): Container metadata
        public_access (PublicAccess, optional): Public access level
        
    Returns:
        ContainerClient: Client for the created container
    """

def delete_container(self, container: str, **kwargs) -> None:
    """
    Delete a container.
    
    Args:
        container (str): Container name to delete
    """

def undelete_container(self, deleted_container_name: str, deleted_container_version: str, **kwargs) -> ContainerClient:
    """
    Restore a soft-deleted container.
    
    Args:
        deleted_container_name (str): Name of deleted container
        deleted_container_version (str): Version of deleted container
        
    Returns:
        ContainerClient: Client for the restored container
    """

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

def get_container_client(self, container: str) -> 'ContainerClient':
    """
    Get a ContainerClient for a specific container.
    
    Args:
        container (str): Container name
        
    Returns:
        ContainerClient: Client for container operations
    """

def get_blob_client(self, container: Union[ContainerProperties, str], blob: str, snapshot: Optional[Union[Dict[str, Any], str]] = None, *, version_id: Optional[str] = None) -> BlobClient:
    """
    Get a BlobClient for a specific blob.
    
    Args:
        container (Union[ContainerProperties, str]): Container name or container properties object
        blob (str): Blob name
        snapshot (Optional[Union[Dict[str, Any], str]]): Blob snapshot identifier or snapshot properties dict
        version_id (Optional[str]): Version ID to target a specific version of the blob
        
    Returns:
        BlobClient: Client for blob operations
    """

# Common configuration options
BlobServiceClient(
    account_url="https://account.blob.core.windows.net",
    credential=credential,
    api_version="2021-12-02",              # API version to use
    secondary_hostname="account-secondary.blob.core.windows.net",  # Secondary endpoint
    max_block_size=4*1024*1024,            # 4MB max block size for uploads
    max_single_put_size=64*1024*1024,      # 64MB threshold for chunked uploads
    min_large_block_upload_threshold=4*1024*1024+1,  # Memory efficient threshold
    use_byte_buffer=False,                 # Use byte buffer for block uploads
    max_page_size=4*1024*1024,            # 4MB max page size
    max_single_get_size=32*1024*1024,     # 32MB threshold for chunked downloads  
    max_chunk_get_size=4*1024*1024,       # 4MB chunk size for downloads
    audience="https://storage.azure.com/"  # Azure AD audience
)

class ExponentialRetry:
    def __init__(self, initial_backoff: int = 15, increment_base: int = 3, retry_total: int = 3, retry_to_secondary: bool = False, random_jitter_range: int = 3, **kwargs):
        """
        Exponential retry policy with increasing backoff intervals.
        
        Args:
            initial_backoff (int): Initial backoff interval in seconds (default: 15)
            increment_base (int): Base to increment backoff after first retry (default: 3)
            retry_total (int): Maximum number of retry attempts (default: 3)  
            retry_to_secondary (bool): Whether to retry to secondary endpoint (default: False)
            random_jitter_range (int): Range in seconds for randomizing backoff (default: 3)
        """

class LinearRetry:
    def __init__(self, backoff: int = 15, retry_total: int = 3, retry_to_secondary: bool = False, random_jitter_range: int = 3, **kwargs):
        """
        Linear retry policy with fixed backoff intervals.
        
        Args:
            backoff (int): Fixed backoff interval in seconds between retries (default: 15)
            retry_total (int): Maximum number of retry attempts (default: 3)
            retry_to_secondary (bool): Whether to retry to secondary endpoint (default: False)  
            random_jitter_range (int): Range in seconds for randomizing backoff (default: 3)
        """

from azure.storage.blob import BlobServiceClient, ExponentialRetry

# Configure exponential retry with custom settings
retry_policy = ExponentialRetry(
    initial_backoff=5,
    increment_base=2, 
    retry_total=5,
    random_jitter_range=2
)

client = BlobServiceClient(
    account_url="https://mystorageaccount.blob.core.windows.net",
    credential="account_key_here",
    retry_policy=retry_policy
)

class StorageErrorCode:
    """Error codes returned by Azure Storage services."""
    
    # Authentication and Authorization
    AUTHENTICATION_FAILED: str = "AuthenticationFailed"
    AUTHORIZATION_FAILURE: str = "AuthorizationFailure"
    INSUFFICIENT_ACCOUNT_PERMISSIONS: str = "InsufficientAccountPermissions"
    ACCOUNT_IS_DISABLED: str = "AccountIsDisabled"
    
    # Condition Errors
    CONDITION_NOT_MET: str = "ConditionNotMet"
    CONDITION_HEADERS_NOT_SUPPORTED: str = "ConditionHeadersNotSupported"
    
    # Resource Errors
    RESOURCE_NOT_FOUND: str = "ResourceNotFound"
    RESOURCE_ALREADY_EXISTS: str = "ResourceAlreadyExists"
    CONTAINER_NOT_FOUND: str = "ContainerNotFound"
    BLOB_NOT_FOUND: str = "BlobNotFound"
    
    # Request Errors
    INVALID_RANGE: str = "InvalidRange"
    INVALID_INPUT: str = "InvalidInput"
    INVALID_METADATA: str = "InvalidMetadata"
    INVALID_HEADER_VALUE: str = "InvalidHeaderValue"
    MD5_MISMATCH: str = "Md5Mismatch"
    
    # Server Errors
    INTERNAL_ERROR: str = "InternalError"
    SERVER_BUSY: str = "ServerBusy"

class PartialBatchErrorException(HttpResponseError):
    """Exception raised when there is a partial failure in batch operations."""
    
    def __init__(self, message: str, response, parts: list):
        """
        Initialize PartialBatchErrorException.
        
        Args:
            message (str): Error message describing the exception
            response: HTTP response object from the failed request
            parts (list): List of the parts in multipart response that failed
        """
        self.parts = parts

from azure.storage.blob import BlobServiceClient, StorageErrorCode
from azure.core.exceptions import HttpResponseError

try:
    client = BlobServiceClient.from_connection_string(conn_str)
    blob_client = client.get_blob_client("container", "nonexistent-blob") 
    properties = blob_client.get_blob_properties()
except HttpResponseError as e:
    if e.error_code == StorageErrorCode.BLOB_NOT_FOUND:
        print("Blob does not exist")
    elif e.error_code == StorageErrorCode.CONTAINER_NOT_FOUND:
        print("Container does not exist")
    else:
        print(f"Unexpected error: {e.error_code}")

class BlobAnalyticsLogging:
    """Analytics logging configuration."""
    version: str
    delete: bool
    read: bool  
    write: bool
    retention_policy: RetentionPolicy

class Metrics:
    """Service metrics configuration."""  
    version: str
    enabled: bool
    include_apis: bool
    retention_policy: RetentionPolicy

class RetentionPolicy:
    """Data retention policy."""
    enabled: bool
    days: int

class StaticWebsite:
    """Static website hosting configuration."""
    enabled: bool
    index_document: str
    error_document404_path: str
    default_index_document_path: str

class CorsRule:
    """Cross-origin resource sharing rule."""
    allowed_origins: list[str]
    allowed_methods: list[str]  
    allowed_headers: list[str]
    exposed_headers: list[str]
    max_age_in_seconds: int

class ContainerProperties:
    """Container properties returned by list operations."""
    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 UserDelegationKey:
    """User delegation key for SAS generation."""
    signed_oid: str
    signed_tid: str  
    signed_start: datetime
    signed_expiry: datetime
    signed_service: str
    signed_version: str
    value: str

class FilteredBlob:
    """Blob found by tag filtering."""
    name: str
    container_name: str
    tag_value: str