Pub/Sub in Google Cloud: A Hands-On Tutorial with gcloud and Python

Imagine a busy post office. Senders drop letters into a box without knowing who will read them. The post office sorts them, and every subscriber to that kind of mail gets their own copy, delivered whenever they are ready to receive it. Nobody waits for anybody else, and if a recipient is on holiday, the letters simply wait.

Google Cloud Pub/Sub is that post office for software. It lets one part of your system announce that "something happened" (a PDF was uploaded, an order was placed, a sensor sent a reading) while other parts react to it independently. In this tutorial we will build the idea from the ground up: first the vocabulary, then gcloud commands to create and use a topic, then Python code to publish and receive messages, and finally a complete event-driven pipeline: Cloud Storage → Pub/Sub → Cloud Run → BigQuery, as in an insurance document-ingestion system.

1. The five words you need to know

Pub/Sub has a small vocabulary. Once these terms are clear, everything else is detail.

• Message: the unit of data: a payload (bytes, usually JSON) plus optional key/value attributes.
• Topic: a named channel to which publishers send messages. Think of it as the mailbox.
• Publisher: any application that sends messages to a topic.
• Subscription: a named connection between a topic and a consumer. It tells Pub/Sub where and how messages must be delivered.
• Subscriber: the application that receives messages from a subscription and acknowledges (acks) them once processed.

Figure 1: Publishers know only the topic. Each subscription receives its own copy of every message.

Two properties of this design explain why Pub/Sub is so useful. First, decoupling: the publisher never needs to know who is listening, so you can add a new consumer later without touching the producer. Second, fan-out: if three subscriptions are attached to one topic, each receives every message. But if several workers share one subscription, they split the messages between them, which is how you scale processing horizontally.

2. Your first topic with gcloud

Open Cloud Shell, or a terminal where gcloud is installed and authenticated. The commands below use Bash syntax; in Windows PowerShell, put each command on one line or replace the trailing \ with a backtick.

# Set your project and enable the APIs we will use
gcloud config set project YOUR_PROJECT_ID
gcloud services enable pubsub.googleapis.com run.googleapis.com \
  bigquery.googleapis.com storage.googleapis.com

# Create a topic
gcloud pubsub topics create policy-files

# Create a pull subscription attached to the topic
gcloud pubsub subscriptions create policy-pull-sub \
  --topic=policy-files \
  --ack-deadline=60

The first block selects the project and enables the APIs. The topic is just a name; it stores nothing by itself. The subscription is what makes messages wait for you: from the moment it exists, Pub/Sub keeps a copy of every new message for it until the message is acknowledged or expires. (Messages published before a subscription existed are not delivered to it, a classic first-time surprise.) The --ack-deadline=60 option gives a subscriber 60 seconds to acknowledge a message before Pub/Sub assumes it failed and redelivers it.

Now publish a message and read it back:

gcloud pubsub topics publish policy-files \
  --message='{"policy_id":"POL-123","premium":1250.50}' \
  --attribute=source=tutorial

gcloud pubsub subscriptions pull policy-pull-sub --limit=5 --auto-ack

The publish command returns a message ID, which confirms that Pub/Sub has stored the message. The pull command asks the subscription for up to five messages; --auto-ack acknowledges them immediately so that they are not delivered again. Without that flag, the messages would reappear after the ack deadline, which is a handy way to see redelivery with your own eyes.

3. Publishing and receiving with Python

The CLI is perfect for exploring, but real applications talk to Pub/Sub through a client library. In this section we will write two small programs, a publisher and a subscriber, and go through them line by line, because the behavior of the library (futures, threads, acknowledgments) is where most beginners get confused.

3.1 Setup: library, credentials, permissions

pip install google-cloud-pubsub
gcloud auth application-default login

The first command installs the official client. The second stores Application Default Credentials on your machine: the Python library finds them automatically, so no key file or password appears in your code. On Cloud Run or Compute Engine, the same code instead uses the service account attached to the service, again without any change.

The identity running the code also needs permission: roles/pubsub.publisher on the topic to publish, and roles/pubsub.subscriber on the subscription to receive. As project owner in a tutorial project you already have both; in production you grant them explicitly.

3.2 The publisher, step by step

import json
from google.cloud import pubsub_v1

PROJECT_ID = "YOUR_PROJECT_ID"

publisher = pubsub_v1.PublisherClient()
topic_path = publisher.topic_path(PROJECT_ID, "policy-files")

payload = {"policy_id": "POL-124", "premium": 980.00}

future = publisher.publish(
    topic_path,
    json.dumps(payload).encode("utf-8"),   # data must be bytes
    source="python-demo",                  # extra kwargs become attributes
)
print("Published message ID:", future.result())

Let us walk through it:

• PublisherClient() creates the connection object. Create it once and reuse it: it is thread-safe and holds the batching machinery described below.
• topic_path(...) builds the full resource name, projects/YOUR_PROJECT_ID/topics/policy-files. Pub/Sub always identifies a topic by this long name; the helper simply saves you from typing it by hand.
• json.dumps(payload).encode("utf-8"): a Pub/Sub message body is raw bytes, not a Python dictionary. JSON is the usual convention because every language can read it.
• source="python-demo": any extra keyword argument becomes a message attribute, a small string key/value pair travelling beside the payload. Attributes are ideal for routing or filtering (a subscription can filter on them) without decoding the body. Both keys and values must be strings.
• future.result(): publish() does not wait for the server. It returns immediately with a future, a promise of a result. Calling result() blocks until Pub/Sub confirms that the message is stored, and returns the message ID. If publishing failed, result() raises the exception, which is how you learn about errors.

Figure 2: What happens when you call publish(). The call returns at once; the actual network send happens in the background.

Why is publish() asynchronous? Because the client batches messages: by default it groups up to 100 messages (or about 1 MB, or a few milliseconds of waiting) into a single network request. This is what lets one process publish thousands of messages per second. The consequence is that you should not call result() after every single message in a loop, since that forces one round trip each time. Instead, collect the futures and check them at the end:

from concurrent.futures import as_completed

futures = []
for i in range(20):
    data = json.dumps({"policy_id": f"POL-{i:03d}"}).encode("utf-8")
    futures.append(publisher.publish(topic_path, data, source="batch-demo"))

for f in as_completed(futures):
    print("Published:", f.result())

Here the twenty publish() calls return instantly and the library sends them in one or two batches. The final loop waits for each confirmation as it arrives and prints the IDs.

3.3 The pull subscriber, step by step

from concurrent.futures import TimeoutError
from google.cloud import pubsub_v1

PROJECT_ID = "YOUR_PROJECT_ID"

subscriber = pubsub_v1.SubscriberClient()
sub_path = subscriber.subscription_path(PROJECT_ID, "policy-pull-sub")

def callback(message: pubsub_v1.subscriber.message.Message) -> None:
    print("Received:", message.data.decode("utf-8"))
    print("Attributes:", dict(message.attributes))
    message.ack()          # success: do not send it again
    # message.nack()       # failure: redeliver as soon as possible

streaming_pull = subscriber.subscribe(sub_path, callback=callback)
print("Listening... (30 seconds)")

with subscriber:
    try:
        streaming_pull.result(timeout=30)
    except TimeoutError:
        streaming_pull.cancel()
        streaming_pull.result()

This program has three parts: the callback, the subscription, and the waiting logic.

The callback. This is your processing code. The library calls it once for every message and hands it a Message object with these useful fields:

Field	Meaning
message.data	The payload, as bytes. Decode it and parse the JSON.
message.attributes	The string key/value pairs set by the publisher (here, source).
message.message_id	The unique ID assigned by Pub/Sub; handy for logs and deduplication.
message.publish_time	When Pub/Sub received the message.
message.delivery_attempt	How many times it has been delivered; available when a dead-letter policy is configured (see Section 5).

The last act of the callback decides the message's fate. ack() says "done, forget it"; nack() says "I could not handle it, send it again soon". If you do neither, Pub/Sub waits for the acknowledgment deadline and then redelivers. One practical rule: do not let an exception escape silently. Wrap your processing in try/except and call nack() explicitly in the except branch, so that the failure is intentional and visible in your logs.

The subscription call. subscribe() opens a streaming pull: one long-lived connection over which Pub/Sub pushes messages to the client as they arrive. Crucially, subscribe() returns immediately: the work happens on background threads, and your callback is run on a thread pool, so several messages are processed in parallel. What you get back is a StreamingPullFuture, a handle on that background activity.

The waiting logic. Since everything runs in the background, the main thread has nothing to do, and a script that simply ended would stop listening at once. Calling streaming_pull.result(timeout=30) blocks the main thread: it waits up to 30 seconds, then raises TimeoutError. In the except branch, cancel() asks the stream to stop, and the following result() waits until shutdown completes cleanly. The with subscriber: block closes the underlying connection on exit. In a real service you would omit the timeout and let the process run until it is stopped.

Figure 3: Streaming pull. Messages flow into a thread pool where your callback runs in parallel; each callback ends with an ack or a nack.

3.4 Two behaviors worth knowing

Automatic lease extension. Remember the 60-second acknowledgment deadline we set earlier? You might worry that a slow callback (say, parsing a large PDF for two minutes) would be redelivered mid-processing. The library prevents this: while your callback is still running, the client keeps extending the lease on the message in the background, up to a configurable maximum. The deadline therefore matters mainly when your process crashes or hangs, since then nobody extends the lease and Pub/Sub rightly assumes the message needs a new owner.

Flow control. By default the client may pull many messages into memory at once. If each message is heavy (a PDF to parse), you want to cap how many are in flight:

flow_control = pubsub_v1.types.FlowControl(max_messages=5)

streaming_pull = subscriber.subscribe(
    sub_path, callback=callback, flow_control=flow_control
)

With max_messages=5, at most five messages are being processed at any moment; the rest wait in the subscription until a callback finishes. This protects your memory and any downstream service (a database, an LLM API) from being flooded.

3.5 Run it and watch

Open two terminals. In the first, start the subscriber; in the second, run the publisher. You should see something like this in the first terminal:

Listening... (30 seconds)
Received: {"policy_id": "POL-124", "premium": 980.0}
Attributes: {'source': 'python-demo'}

Now try two experiments to make the mechanics concrete. First, comment out message.ack() and run again: the same message reappears after the acknowledgment deadline, proving redelivery. Second, stop the subscriber, publish three messages, then restart it: all three arrive at once, because the subscription kept them while nobody was listening. Remember that Pub/Sub does not guarantee ordering by default, so do not assume messages arrive in the order you published them.

4. Pull or push? Two ways to deliver

So far the subscriber went to Pub/Sub to get its messages. That is pull. Pub/Sub can also do the opposite: it can call your HTTPS endpoint whenever a message arrives. That is push, and it is a natural fit for Cloud Run, which has no long-running process to hold a connection open.

Figure 4: With pull, the subscriber fetches. With push, Pub/Sub calls an HTTP endpoint and treats a 2xx response as the acknowledgment.

Receiving push messages in Cloud Run

A push subscription needs a service to call. Here is a minimal Flask handler. Notice what is missing: there is no SubscriberClient and no polling loop. Pub/Sub simply POSTs to the endpoint.

import base64
import json
from flask import Flask, request

app = Flask(__name__)

@app.post("/")
def handle_pubsub():
    envelope = request.get_json(silent=True)
    if not envelope or "message" not in envelope:
        return "Bad Request: not a Pub/Sub message", 400

    message = envelope["message"]
    data = json.loads(base64.b64decode(message["data"]).decode("utf-8"))
    print("Message ID:", message["messageId"], "Data:", data)

    # ... do the real work here ...

    return "", 204      # 2xx = acknowledged; anything else = retry

The push request wraps your message in a JSON envelope, and the data field is Base64-encoded, so the code decodes it first. Returning 204 acknowledges the message; returning an error code (or timing out) tells Pub/Sub to try again.

After deploying the service, create the push subscription. It must authenticate to Cloud Run, so we first create a dedicated service account, allow it to invoke the service, and attach it to the subscription:

# Deploy the Flask app (requirements.txt: flask, gunicorn)
gcloud run deploy policy-processor \
  --source . --region=europe-west1 --no-allow-unauthenticated

# Identity that Pub/Sub will use to call Cloud Run
gcloud iam service-accounts create pubsub-invoker

gcloud run services add-iam-policy-binding policy-processor \
  --region=europe-west1 \
  --member="serviceAccount:pubsub-invoker@YOUR_PROJECT_ID.iam.gserviceaccount.com" \
  --role="roles/run.invoker"

# The push subscription
gcloud pubsub subscriptions create policy-subscription \
  --topic=policy-files \
  --push-endpoint="https://YOUR-CLOUD-RUN-URL/" \
  --push-auth-service-account="pubsub-invoker@YOUR_PROJECT_ID.iam.gserviceaccount.com"

Read the last command as a sentence: "Whenever a message arrives on policy-files, send an authenticated HTTP POST to this URL, signing the request as pubsub-invoker." The subscription lives in Pub/Sub, not in your Python code, which is why it is created with gcloud (or Terraform, or the console).

5. What happens when things go wrong

The most important promise Pub/Sub makes is at-least-once delivery: a message stays in the subscription until it is acknowledged. If your code fails, times out, or returns a 5xx, the message comes back. This is wonderful for reliability, but it raises two questions: what if a message never succeeds, and what if it is delivered twice?

Figure 5: A message is delivered until it is acknowledged. Messages that keep failing can be parked in a dead-letter topic.

Dead-letter topics: a place for poison messages

A "poison" message (say, a corrupted PDF) would otherwise be retried forever. A dead-letter topic gives it somewhere to go after a fixed number of attempts, so that you can inspect it later without blocking the rest of the traffic.

gcloud pubsub topics create policy-files-dead-letter
gcloud pubsub subscriptions create policy-dead-letter-sub \
  --topic=policy-files-dead-letter

gcloud pubsub subscriptions update policy-subscription \
  --dead-letter-topic=policy-files-dead-letter \
  --max-delivery-attempts=5 \
  --min-retry-delay=10s --max-retry-delay=600s

# Let the Pub/Sub service agent move failed messages
PROJECT_NUMBER=$(gcloud projects describe YOUR_PROJECT_ID --format="value(projectNumber)")
PUBSUB_SA="service-${PROJECT_NUMBER}@gcp-sa-pubsub.iam.gserviceaccount.com"

gcloud pubsub topics add-iam-policy-binding policy-files-dead-letter \
  --member="serviceAccount:${PUBSUB_SA}" --role="roles/pubsub.publisher"
gcloud pubsub subscriptions add-iam-policy-binding policy-subscription \
  --member="serviceAccount:${PUBSUB_SA}" --role="roles/pubsub.subscriber"

The --min-retry-delay and --max-retry-delay options create an exponential backoff, so a struggling downstream system is not hammered. The two IAM bindings matter: Pub/Sub itself must be allowed to publish to the dead-letter topic and to acknowledge messages on the original subscription. Forgetting them is the most common reason a dead-letter setup silently does nothing. Also create a subscription on the dead-letter topic (as we did), because a topic without subscriptions discards its messages.

Duplicates: design for idempotency

Because delivery is at-least-once, your handler may occasionally see the same message twice. The fix is not to prevent duplicates, but to make them harmless: processing the same message twice must give the same result as processing it once. In a file pipeline, a natural unique key is the object's bucket, name, and generation:

row_id = f"{bucket}/{name}#{generation}"
errors = bq_client.insert_rows_json(TABLE_ID, [policy], row_ids=[row_id])

Here row_ids lets BigQuery's streaming insert discard repeats on a best-effort basis. For strict guarantees, use a MERGE statement keyed on the same identifier.

6. A complete pipeline: Cloud Storage to BigQuery

Now we assemble everything. An insurance company uploads policy PDFs to a bucket. Each upload should automatically produce a row in BigQuery.

Figure 6: The end-to-end event-driven ingestion pipeline.

Step 1: connect the bucket to the topic

gcloud storage buckets create gs://YOUR_BUCKET --location=europe-west1

# Allow Cloud Storage to publish to the topic
SERVICE_AGENT=$(gcloud storage service-agent --project=YOUR_PROJECT_ID)
gcloud pubsub topics add-iam-policy-binding policy-files \
  --member="serviceAccount:${SERVICE_AGENT}" --role="roles/pubsub.publisher"

# Publish an event for every new object
gcloud storage buckets notifications create gs://YOUR_BUCKET \
  --topic=policy-files \
  --event-types=OBJECT_FINALIZE \
  --payload-format=json

OBJECT_FINALIZE fires when an upload completes. With --payload-format=json, the message body is the object's metadata, which includes the bucket and name fields our handler reads.

Step 2: the Cloud Run handler

import base64
import json

from flask import Flask, request
from google.cloud import bigquery, storage

app = Flask(__name__)
storage_client = storage.Client()
bq_client = bigquery.Client()

TABLE_ID = "YOUR_PROJECT_ID.insurance.policies"

@app.post("/")
def handle_pubsub():
    envelope = request.get_json(silent=True)
    if not envelope or "message" not in envelope:
        return "Bad Request", 400

    event = json.loads(base64.b64decode(envelope["message"]["data"]))
    bucket, name = event["bucket"], event["name"]
    generation = event.get("generation", "0")

    # 1. Download the PDF
    pdf_bytes = storage_client.bucket(bucket).blob(name).download_as_bytes()

    # 2. Extract information (placeholder: plug in your real parser here)
    policy = {"policy_id": "POL-123", "customer_id": "CUST-456", "premium": 1250.50}

    # 3. Load into BigQuery, with a stable ID against duplicates
    errors = bq_client.insert_rows_json(
        TABLE_ID, [policy], row_ids=[f"{bucket}/{name}#{generation}"]
    )
    if errors:
        print("BigQuery errors:", errors)
        return "BigQuery insert failed", 500   # Pub/Sub will retry

    return "", 204   # success: acknowledged

The order of the final lines is deliberate. We return a success code only after BigQuery accepted the row. If BigQuery fails temporarily, the 500 makes Pub/Sub retry, and the event is not lost. Create the table beforehand:

bq mk --dataset YOUR_PROJECT_ID:insurance
bq mk --table YOUR_PROJECT_ID:insurance.policies \
  policy_id:STRING,customer_id:STRING,premium:FLOAT

Step 3: test it

gcloud storage cp sample-policy.pdf gs://YOUR_BUCKET/incoming/
bq query --use_legacy_sql=false \
  'SELECT * FROM `YOUR_PROJECT_ID.insurance.policies`'

If no row appears, look at the Cloud Run logs and at the subscription's undelivered-message count: that single number tells you quickly whether the problem is before Pub/Sub (the notification) or after it (your handler).

7. Who is allowed to do what? Three identities

IAM is where most first attempts fail, mainly because three different identities are involved, each with its own job.

Identity	Purpose	Permission
Cloud Storage service agent	Publishes file-created notifications	roles/pubsub.publisher on the topic
pubsub-invoker service account	Authenticates push requests to Cloud Run	roles/run.invoker on the service
Cloud Run runtime service account	Reads PDFs and writes results	roles/storage.objectViewer on the bucket, roles/bigquery.dataEditor on the dataset

The invoker account only opens the door to Cloud Run. It does not become the identity your Python code uses to talk to Cloud Storage or BigQuery; that is the runtime service account's role.

8. Pub/Sub, Eventarc, Workflows, Dataflow: choosing wisely

Pub/Sub is the messaging layer, and it combines with other services that are easy to confuse with it:

• Eventarc routes Cloud Storage (and other) events to Cloud Run and manages the Pub/Sub plumbing for you. It is simpler in production, but building the Pub/Sub version first, as we did, teaches you the topic, subscription, acknowledgment, and IAM concepts that Eventarc hides.
• Workflows orchestrates several services in sequence, with retries and conditions. It decides what runs next.
• Dataflow processes huge streams or batches of data in parallel (for example, Pub/Sub → Dataflow → BigQuery for millions of records per day). It decides how data is processed at scale.

Rule of thumb: Pub/Sub carries events, Cloud Run executes application logic, Workflows coordinates steps, Dataflow scales data transformations, and BigQuery stores the results.

9. Clean up

gcloud pubsub subscriptions delete policy-subscription policy-pull-sub policy-dead-letter-sub
gcloud pubsub topics delete policy-files policy-files-dead-letter
gcloud run services delete policy-processor --region=europe-west1
gcloud storage rm --recursive gs://YOUR_BUCKET
bq rm -r -f -d YOUR_PROJECT_ID:insurance

Remember: deleting a topic does not delete its subscriptions; they become orphaned. Delete subscriptions first, as above.

Summary

A topic receives messages; a subscription decides how and where a copy is delivered; a subscriber acknowledges success. Pull lets your code fetch at its own pace, while push lets Pub/Sub call a Cloud Run endpoint, where a 2xx response is the acknowledgment. Delivery is at-least-once, so combine retries, a dead-letter topic, and idempotent processing. With these ideas, a few gcloud commands and a short Flask app are enough to turn a file upload into a BigQuery row, reliably and without a single polling loop.

Official references: Pub/Sub overview, Cloud Run Pub/Sub tutorial, Push subscriptions, Cloud Storage notifications.