Azure Blob Storage browser - list and download blobs with filtering and paging

list_blobs.py

#!/usr/bin/env python3
"""List blobs in an Azure Storage container with paging, or download a specific blob."""

import argparse
import os
import sys
from dotenv import load_dotenv
from azure.identity import ClientSecretCredential
from azure.storage.blob import BlobServiceClient

load_dotenv()

# Load credentials from environment
tenant_id = os.getenv("TENANT_ID")
client_id = os.getenv("CLIENT_ID")
client_secret = os.getenv("CLIENT_SECRET")
storage_account = os.getenv("STORAGE_ACCOUNT")
container_name = os.getenv("CONTAINER_NAME")

def check_env():
    if not all([tenant_id, client_id, client_secret, storage_account]):
        print("Error: Missing environment variables. Check your .env file.")
        print(f"  TENANT_ID: {'set' if tenant_id else 'MISSING'}")
        print(f"  CLIENT_ID: {'set' if client_id else 'MISSING'}")
        print(f"  CLIENT_SECRET: {'set' if client_secret else 'MISSING'}")
        print(f"  STORAGE_ACCOUNT: {'set' if storage_account else 'MISSING'}")
        exit(1)

def get_blob_service_client():
    credential = ClientSecretCredential(
        tenant_id=tenant_id,
        client_id=client_id,
        client_secret=client_secret
    )
    account_url = f"https://{storage_account}.blob.core.windows.net"
    return BlobServiceClient(account_url, credential=credential)

def list_blobs(container, page_size, filter_str=None):
    """List blobs with paging, most recent first."""
    blob_service_client = get_blob_service_client()
    container_client = blob_service_client.get_container_client(container)

    print(f"Fetching blobs from '{container}'...\n")

    # Get all blobs and sort by last_modified descending
    blobs = list(container_client.list_blobs())
    blobs.sort(key=lambda b: b.last_modified, reverse=True)

    # Apply filter if specified
    if filter_str:
        blobs = [b for b in blobs if filter_str.lower() in b.name.lower()]
        print(f"Filtered by '{filter_str}': {len(blobs)} matching blobs.\n")

    if not blobs:
        print("No blobs found in this container.")
        return

    print(f"Found {len(blobs)} blobs. Showing most recent first.\n")

    # Page through results
    idx = 0
    is_tty = sys.stdout.isatty()
    first_page = True

    while idx < len(blobs):
        page = blobs[idx:idx + page_size]

        # Print header: once in pipe mode, every page in TTY mode
        if first_page or is_tty:
            print(f"{'#':<6} {'Last Modified':<22} {'Size':<12} {'Name'}")
            print("-" * 100)
            first_page = False

        for i, blob in enumerate(page):
            num = idx + i + 1
            modified = blob.last_modified.strftime("%Y-%m-%d %H:%M:%S")
            size = f"{blob.size:,} B"
            print(f"{num:<6} {modified:<22} {size:<12} {blob.name}")

        idx += page_size

        if idx < len(blobs) and is_tty:
            print(f"\n--- Showing {idx}/{len(blobs)} blobs. Press Enter for next page, 'q' to quit ---")
            choice = input().strip().lower()
            if choice == 'q':
                break
            print()

def download_blob(container, blob_name, output_file=None):
    """Download a specific blob."""
    blob_service_client = get_blob_service_client()
    container_client = blob_service_client.get_container_client(container)
    blob_client = container_client.get_blob_client(blob_name)

    if output_file is None:
        output_file = os.path.basename(blob_name)

    print(f"Downloading '{blob_name}' to '{output_file}'...")

    with open(output_file, "wb") as f:
        data = blob_client.download_blob()
        f.write(data.readall())

    print(f"Done. Saved to '{output_file}'")

def main():
    parser = argparse.ArgumentParser(description="List or download blobs from Azure Storage")
    parser.add_argument("-c", "--container", default=container_name,
                        help="Container name (default: from CONTAINER_NAME env var)")
    parser.add_argument("-p", "--page-size", type=int, default=20,
                        help="Number of blobs per page (default: 20)")
    parser.add_argument("-d", "--download", metavar="BLOB_NAME",
                        help="Download a specific blob by name")
    parser.add_argument("-o", "--output", metavar="FILE",
                        help="Output filename for download (default: blob basename)")
    parser.add_argument("-f", "--filter", metavar="STRING",
                        help="Filter blob names containing STRING (case-insensitive)")

    args = parser.parse_args()

    check_env()

    if not args.container:
        print("Error: No container specified. Use -c or set CONTAINER_NAME in .env")
        exit(1)

    if args.download:
        download_blob(args.container, args.download, args.output)
    else:
        list_blobs(args.container, args.page_size, args.filter)

if __name__ == "__main__":
    main()

requirements.txt

azure-identity
azure-storage-blob
python-dotenv