Bucket Operations

def __init__(self, storage, name):
    """
    Initialize a Bucket instance.
    
    Parameters:
    - storage (Storage): Parent storage client instance
    - name (str): Bucket name
    
    Attributes:
    - storage (Storage): Reference to parent storage client
    - name (str): Bucket name
    """

async with Storage() as storage:
    # Get bucket instance
    bucket = storage.get_bucket('my-bucket')
    
    # Bucket properties
    print(f"Bucket name: {bucket.name}")
    print(f"Storage client: {bucket.storage}")

async def get_blob(self, blob_name, timeout=10, session=None):
    """
    Get a blob instance with metadata loaded.
    
    Parameters:
    - blob_name (str): Name/path of the blob
    - timeout (int): Request timeout in seconds
    - session (aiohttp.ClientSession, optional): Custom session
    
    Returns:
    Blob: Blob instance with loaded metadata
    """

async def blob_exists(self, blob_name, session=None):
    """
    Check if a blob exists in the bucket.
    
    Parameters:
    - blob_name (str): Name/path of the blob to check
    - session (aiohttp.ClientSession, optional): Custom session
    
    Returns:
    bool: True if blob exists, False otherwise
    """

def new_blob(self, blob_name):
    """
    Create a new blob instance without metadata.
    
    Parameters:
    - blob_name (str): Name/path for the new blob
    
    Returns:
    Blob: New blob instance (not yet uploaded)
    """

async with Storage() as storage:
    bucket = storage.get_bucket('my-bucket')
    
    # Check if blob exists
    if await bucket.blob_exists('data.json'):
        # Get existing blob with metadata
        blob = await bucket.get_blob('data.json')
        print(f"Blob size: {blob.size} bytes")
    else:
        # Create new blob instance
        blob = bucket.new_blob('data.json')
        # Upload data to the new blob
        await blob.upload(json_data, content_type='application/json')

async def list_blobs(self, prefix='', match_glob='', delimiter='', session=None):
    """
    List blob names in the bucket.
    
    Parameters:
    - prefix (str): Only return blobs whose names begin with this prefix
    - match_glob (str): Glob pattern to match blob names against
    - delimiter (str): Used to simulate a folder structure (e.g., '/')
    - session (aiohttp.ClientSession, optional): Custom session
    
    Returns:
    List[str]: List of blob names matching the criteria
    """

async with Storage() as storage:
    bucket = storage.get_bucket('my-bucket')
    
    # List all blobs
    all_blobs = await bucket.list_blobs()
    
    # List blobs with prefix (folder-like structure)
    log_files = await bucket.list_blobs(prefix='logs/')
    
    # List blobs matching glob pattern
    json_files = await bucket.list_blobs(match_glob='*.json')
    
    # List with delimiter for folder simulation
    top_level = await bucket.list_blobs(delimiter='/')
    
    # Combine filters
    recent_logs = await bucket.list_blobs(prefix='logs/2023/', match_glob='*.log')

async def get_metadata(self, params=None, session=None):
    """
    Get bucket metadata and configuration.
    
    Parameters:
    - params (dict, optional): Additional query parameters
    - session (aiohttp.ClientSession, optional): Custom session
    
    Returns:
    Dict[str, Any]: Bucket metadata including creation time, location, storage class, etc.
    """

async with Storage() as storage:
    bucket = storage.get_bucket('my-bucket')
    
    # Get bucket metadata
    metadata = await bucket.get_metadata()
    
    print(f"Bucket location: {metadata.get('location')}")
    print(f"Storage class: {metadata.get('storageClass')}")
    print(f"Created: {metadata.get('timeCreated')}")
    print(f"Versioning enabled: {metadata.get('versioning', {}).get('enabled', False)}")

async with Storage() as storage:
    bucket = storage.get_bucket('my-bucket')
    
    # Get all JSON files and process them
    json_files = await bucket.list_blobs(match_glob='*.json')
    
    # Process each file
    for blob_name in json_files:
        blob = await bucket.get_blob(blob_name)
        content = await blob.download()
        # Process content
        
        # Update metadata after processing
        metadata = {'processed': 'true', 'processed_at': datetime.now().isoformat()}
        await storage.patch_metadata(bucket.name, blob_name, metadata)

async with Storage() as storage:
    bucket = storage.get_bucket('my-data-bucket')
    
    # List top-level "folders"
    folders = await bucket.list_blobs(delimiter='/')
    
    # List files in a specific "folder"
    files_in_folder = await bucket.list_blobs(prefix='images/')
    
    # Create folder-like structure when uploading
    for i, image_data in enumerate(image_files):
        blob = bucket.new_blob(f'images/batch-{batch_id}/image-{i}.jpg')
        await blob.upload(image_data, content_type='image/jpeg')

async with Storage() as storage:
    bucket = storage.get_bucket('my-bucket')
    
    # Check before creating to avoid overwriting
    blob_name = 'important-data.json'
    if not await bucket.blob_exists(blob_name):
        blob = bucket.new_blob(blob_name)
        await blob.upload(data, content_type='application/json')
        print("Data uploaded successfully")
    else:
        print("File already exists, skipping upload")
        
    # Or get existing and update
    if await bucket.blob_exists(blob_name):
        blob = await bucket.get_blob(blob_name)
        # Update existing blob
        await blob.upload(updated_data, content_type='application/json')