SDK examples

To run these clients in continuous integration, follow LocalCloud in GitHub Actions for service selection and job-scoped SDK configuration.

LocalCloud accepts selected Google Cloud SDK workflows, but compatibility is service- and operation-specific. Start the runtime and load the endpoint values generated for the selected data volume before running any example:

localcloud start --local-only
eval "$(localcloud env)"

Do not hard-code default ports when using the host CLI: if a default port is already in use, the CLI picks another one. Run examples in the same shell or inject the generated variables into the test process. If a client tries to contact real Google Cloud, stop rather than adding real credentials.

Cloud Storage: Python

This example is repeatable against a persistent local volume:

import os
from google.api_core.exceptions import NotFound
from google.auth.credentials import AnonymousCredentials
from google.cloud import storage

assert os.environ['STORAGE_EMULATOR_HOST']
client = storage.Client(
    project=os.environ['GOOGLE_CLOUD_PROJECT'],
    credentials=AnonymousCredentials(),
)
bucket = client.bucket("localcloud-sdk-example")

try:
    bucket.reload()
except NotFound:
    bucket.create()

blob = bucket.blob("data.txt")
blob.upload_from_string("Hello from LocalCloud!")
print(blob.download_as_text())

Cloud Storage: Node.js

const { Storage } = require('@google-cloud/storage');

if (!process.env.STORAGE_EMULATOR_HOST || !process.env.GOOGLE_CLOUD_PROJECT) {
  throw new Error('Export the LocalCloud environment before running this example.');
}
const storage = new Storage({ projectId: process.env.GOOGLE_CLOUD_PROJECT });
const bucket = storage.bucket('localcloud-sdk-example');

async function main() {
  const [exists] = await bucket.exists();
  if (!exists) await storage.createBucket(bucket.name);
  await bucket.file('data.txt').save('Hello from LocalCloud!');
  const [content] = await bucket.file('data.txt').download();
  console.log(content.toString());
}

main().catch((error) => {
  console.error(error);
  process.exitCode = 1;
});

Pub/Sub: Python

Create unique test resources so repeated runs receive their own message, then delete the topic and subscription.

import os
from uuid import uuid4
from google.cloud import pubsub_v1

assert os.environ['PUBSUB_EMULATOR_HOST']
project = os.environ['GOOGLE_CLOUD_PROJECT']
publisher = pubsub_v1.PublisherClient()
subscriber = pubsub_v1.SubscriberClient()
name = 'hello-' + uuid4().hex[:12]
topic = publisher.topic_path(project, name)
subscription = subscriber.subscription_path(project, name)

publisher.create_topic(request={'name': topic}, timeout=30)
try:
    subscriber.create_subscription(request={'name': subscription, 'topic': topic}, timeout=30)
    try:
        publisher.publish(topic, b'Hello, Pub/Sub!').result(timeout=30)
        response = subscriber.pull(request={'subscription': subscription, 'max_messages': 1}, timeout=30)
        assert response.received_messages
        for received in response.received_messages:
            print(received.message.data.decode())
            subscriber.acknowledge(request={'subscription': subscription, 'ack_ids': [received.ack_id]}, timeout=30)
    finally:
        subscriber.delete_subscription(request={'subscription': subscription}, timeout=30)
finally:
    publisher.delete_topic(request={'topic': topic}, timeout=30)
    subscriber.close()
    publisher.transport.close()

BigQuery: deterministic query

The Python BigQuery client needs an explicit REST API endpoint. Read the generated variable, then pass it through client_options with anonymous credentials. A missing variable fails before a client is created. Client configuration reference.

import os
from google.auth.credentials import AnonymousCredentials
from google.cloud import bigquery

client = bigquery.Client(
    project=os.environ['GOOGLE_CLOUD_PROJECT'],
    credentials=AnonymousCredentials(),
    client_options={"api_endpoint": os.environ["BIGQUERY_EMULATOR_HOST"]},
)
rows = list(client.query("SELECT 'Hello from LocalCloud!' AS message").result(timeout=30))
assert rows[0].message == 'Hello from LocalCloud!'
print(rows[0].message)
client.close()

See the BigQuery guide and coverage matrix for details on BQML, vector search, RLS, and the Storage Read API.

Bigtable: Python

Create a test-owned instance and table, write a cell, and read the exact bytes back. This example follows the upstream Bigtable integration guide. Review Bigtable workflows and boundaries before running it.

import os
from uuid import uuid4
from google.cloud import bigtable

assert os.environ['BIGTABLE_EMULATOR_HOST']
client = bigtable.Client(project=os.environ['GOOGLE_CLOUD_PROJECT'], admin=True)
instance = client.instance('guide-' + uuid4().hex[:12])
instance.create(clusters=[instance.cluster('local', location_id='us-central1-a', serve_nodes=1)]).result(timeout=60)
try:
    table = instance.table('messages')
    table.create(column_families={'cf': None})
    row = table.direct_row(b'row-1')
    row.set_cell('cf', b'message', b'hello')
    row.commit()
    message = table.read_row(b'row-1').cells['cf'][b'message'][0].value
    assert message == b'hello'
    print(message.decode())
    table.delete()
finally:
    instance.delete()
    client.close()

Spanner: Python

Create a test-owned instance and database, apply DDL, write a row, and query it through the native SDK. Review Spanner workflows and boundaries for the operations and recovery behavior your application needs.

import os
from uuid import uuid4
from google.cloud import spanner

assert os.environ['SPANNER_EMULATOR_HOST']
client = spanner.Client(project=os.environ['GOOGLE_CLOUD_PROJECT'])
instance = client.instance('guide-' + uuid4().hex[:12], configuration_name='projects/' + client.project + '/instanceConfigs/emulator-config', display_name='Guide', node_count=1)
instance.create().result(timeout=60)
try:
    database = instance.database('messages', ddl_statements=['CREATE TABLE Messages (Id INT64 NOT NULL, Message STRING(100)) PRIMARY KEY (Id)'])
    database.create().result(timeout=60)
    with database.batch() as batch:
        batch.insert('Messages', columns=('Id', 'Message'), values=[(1, 'hello')])
    with database.snapshot() as snapshot:
        rows = list(snapshot.execute_sql('SELECT Message FROM Messages WHERE Id = 1'))
        assert rows == [['hello']]
        print(rows[0][0])
    database.drop()
finally:
    instance.delete()
    client.close()

Firestore: Python

Enable Firestore before loading environment variables; it is disabled by default. Wait for readiness in the setup guide, then run the example in the same shell:

localcloud start --local-only --services firestore
eval "$(localcloud env)"

This example creates, reads, updates, and deletes one unique document in the default database. Review Firestore workflows and boundaries before expanding it to transactions or listeners.

import os
from uuid import uuid4
from google.cloud import firestore

assert os.environ['FIRESTORE_EMULATOR_HOST']
client = firestore.Client(project=os.environ['GOOGLE_CLOUD_PROJECT'])
document = client.collection('guide').document(uuid4().hex)
try:
    document.set({'message': 'hello'}, timeout=30)
    message = document.get(timeout=30).get('message')
    assert message == 'hello'
    print(message)
    document.update({'message': 'updated'}, timeout=30)
    assert document.get(timeout=30).get('message') == 'updated'
finally:
    document.delete(timeout=30)
    client.close()

Selecting another project

Generated environment output uses the configured project. Manual-Docker users can select an explicit project in the request:

eval "$(curl -fsS 'http://localhost:5380/env?format=shell&project=my-local-project')"

Keep client project IDs consistent with the generated GOOGLE_CLOUD_PROJECT value.

Production validation

Local checks cover only documented workflows. Before production deployment:

1. Start a clean process without LocalCloud emulator or custom-endpoint variables.
2. Intentionally configure an isolated real Google Cloud project and credentials.
3. Run production-bound integration tests for IAM, quotas, networking, regional behavior, concurrency, recovery, and unsupported operations.

Do not treat protocol similarity or a passing emulator test as full Google Cloud parity. Organization and team-CI use must also comply with the governing proprietary license.