boostedblob

import asyncio import os from boostedblob.path import LocalPath, AzurePath from boostedblob import delete, list_dir from boostedblob.read import read_json from boostedblob.write import write_json async def main(): # 1. Local File Operations print("--- Local File Operations ---") local_path = LocalPath("./boostedblob_quickstart_test.json") try: await write_json({"message": "Hello from BoostedBlob local!"}, local_path) print(f"Wrote to {local_path}") content = await read_json(local_path) print(f"Read from {local_path}: {content}") # Clean up await delete(local_path) print(f"Deleted {local_path}") except Exception as e: print(f"Error during local operations: {e}") # 2. Azure Blob Storage Operations (requires environment variables) print("\n--- Azure Blob Storage Operations ---") azure_account = os.environ.get('AZURE_STORAGE_ACCOUNT_NAME') azure_container = os.environ.get('AZURE_STORAGE_CONTAINER') if azure_account and azure_container: # Example Azure path azure_path = AzurePath(f"az://{azure_account}/{azure_container}/boostedblob_quickstart_azure.json") print(f"Target Azure Path: {azure_path}") try: await write_json({"message": "Hello from BoostedBlob Azure!"}, azure_path) print(f"Wrote to {azure_path}") content = await read_json(azure_path) print(f"Read from {azure_path}: {content}") # Uncomment the line below to delete the file after reading # await delete(azure_path) # print(f"Deleted {azure_path}") print("Azure operations completed. (Deletion is commented out for safety).") except Exception as e: print(f"Azure operations failed (check credentials, account, container, permissions): {e}") else: print("Skipping Azure operations: Set AZURE_STORAGE_ACCOUNT_NAME and AZURE_STORAGE_CONTAINER environment variables to enable.") print("For GCS, install with `boostedblob[gcs]` and set `GOOGLE_APPLICATION_CREDENTIALS` and `GCS_BUCKET_NAME`.") if __name__ == "__main__": asyncio.run(main())