Google Cloud Storage Emulator

2024.8.3 · active · verified Thu Apr 16

The `gcp-storage-emulator` is a Python library that provides a stub emulator for the Google Cloud Storage API, enabling local development and testing without requiring a connection to the actual Google Cloud Storage service. It's actively maintained, with frequent releases, and is currently at version 2024.8.3. This emulator is particularly useful for accelerating development cycles and running integration tests locally.

Common errors

Warnings

Install

Imports

Quickstart

This quickstart demonstrates how to programmatically start the `gcp-storage-emulator`, configure the `google-cloud-storage` client to interact with it, and perform basic operations like creating a bucket, uploading a blob, and downloading its content. The emulator is configured for in-memory storage for ephemeral testing.

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"]

view raw JSON →