gcp-storage-emulator

import os from google.cloud import storage, exceptions from gcp_storage_emulator.server import create_server HOST = "localhost" PORT = 9023 BUCKET_NAME = "my-test-bucket" FILE_NAME = "test-blob.txt" CONTENT = b"Hello, GCS Emulator!" # 1. Start the emulator server # The default_bucket parameter creates the bucket automatically upon server start. server = create_server(HOST, PORT, in_memory=True, default_bucket=BUCKET_NAME) server.start() try: # 2. Configure the Google Cloud Storage client to use the emulator # Use os.environ.setdefault to allow external configuration. os.environ.setdefault("STORAGE_EMULATOR_HOST", f"http://{HOST}:{PORT}") # 3. Create a client. Project ID is arbitrary for the emulator. client = storage.Client(project="test-project") # 4. Get the bucket (created by default_bucket in server or create manually) bucket = client.bucket(BUCKET_NAME) # Ensure the bucket exists (useful if not using default_bucket param) try: bucket.create() print(f"Bucket '{BUCKET_NAME}' created.") except exceptions.Conflict: # Bucket already exists print(f"Bucket '{BUCKET_NAME}' already exists.") pass # 5. Upload a blob blob = bucket.blob(FILE_NAME) blob.upload_from_string(CONTENT) print(f"Uploaded '{FILE_NAME}' with content: {CONTENT.decode()}.") # 6. Download the blob downloaded_content = blob.download_as_bytes() print(f"Downloaded '{FILE_NAME}' with content: {downloaded_content.decode()}.") # 7. List blobs in the bucket print(f"Blobs in '{BUCKET_NAME}':") for listed_blob in bucket.list_blobs(): print(f" - {listed_blob.name}") finally: # 8. Stop the emulator server server.stop() print("GCP Storage Emulator stopped.") # Clean up the environment variable if needed, though usually not critical after stop if "STORAGE_EMULATOR_HOST" in os.environ: del os.environ["STORAGE_EMULATOR_HOST"]