Publishing Events with PubSub

Welcome to your journey into event-driven architecture with Google Cloud Pub/Sub! In this course, you'll master the art of building robust messaging systems that can handle real-world application events efficiently and reliably.

Google Cloud Pub/Sub is a messaging service that enables you to build event-driven systems where different parts of your application can communicate asynchronously. Think of it as a sophisticated postal system for your code — publishers send messages to topics, and subscribers receive those messages when they're ready to process them. This decoupling allows your applications to scale independently and handle failures gracefully.

The Pub/Sub emulator is your best friend during development and testing. Instead of connecting to Google Cloud's actual Pub/Sub service (which requires authentication, billing setup, and internet connectivity), the emulator runs locally on your machine. This means you can develop, test, and debug your messaging logic without any cloud dependencies or costs. It's particularly valuable when preparing for coding assessments, as you can focus entirely on your implementation without worrying about cloud configuration.

Throughout this course, we'll work with a practical scenario that mirrors real-world applications: handling user signup events. When a user registers for your web application, you want to trigger various downstream processes — sending welcome emails, updating analytics, provisioning user resources, and more. Rather than handling all these tasks synchronously (which would slow down the signup process), you'll publish a signup event that other services can process independently.

Before you can start publishing messages, you need to configure your Python environment to connect to the Pub/Sub emulator instead of the actual Google Cloud service. This configuration is surprisingly simple but absolutely critical for local development.

First, make sure the Pub/Sub emulator is running locally. If you're working on your own machine, you can start the emulator using the following command:

# Start the Pub/Sub emulator on localhost:8681
gcloud beta emulators pubsub start --host-port=localhost:8681

If you're using the CodeSignal IDE, the Pub/Sub emulator is already running on localhost:8681, so you do not need to start it manually.

The key to connecting to the emulator is setting the PUBSUB_EMULATOR_HOST environment variable. This tells the Google Cloud client libraries to send requests to your local emulator instead of the cloud service:

import os
os.environ.setdefault("PUBSUB_EMULATOR_HOST", "localhost:8681")
print("Pub/Sub emulator target →", os.environ["PUBSUB_EMULATOR_HOST"])

The setdefault method is particularly useful here because it only sets the environment variable if it doesn't already exist. This allows you to override the default port if needed while providing a sensible fallback. Port 8681 is the standard port for the Pub/Sub emulator, though you can configure it differently when starting the emulator.

When you run this code, you'll see output confirming your emulator target:

Pub/Sub emulator target → localhost:8681

For your local development, you'll also need the Google Cloud Pub/Sub Python client library. You can install it using pip with pip install google-cloud-pubsub, though in CodeSignal environments, this library comes pre-installed, so you don't need to worry about installation here.

The beauty of this setup is that your code remains identical whether you're running against the emulator or the actual cloud service — only the environment variable changes. This makes it easy to develop locally and deploy to production without code modifications.

Now that your environment is configured, you can start working with Pub/Sub concepts. The foundation of any Pub/Sub system consists of publishers (which send messages) and topics (which receive and store messages until subscribers are ready to process them).

Creating a publisher client is straightforward with the Google Cloud library:

from google.cloud import pubsub_v1

PROJECT, TOPIC = "demo-local", "user-signups"
publisher = pubsub_v1.PublisherClient()
tp = publisher.topic_path(PROJECT, TOPIC)

The PublisherClient automatically detects your emulator configuration from the environment variable you set earlier. The topic_path method creates a properly formatted topic identifier that follows Google Cloud's naming conventions. This path combines your project ID and topic name into a format like projects/demo-local/topics/user-signups.

Creating the topic itself requires a bit of defensive programming because you might run your code multiple times:

try:
    publisher.create_topic(name=tp)
except Exception:
    pass

This approach attempts to create the topic and silently ignores any errors. While this might seem crude, it's actually a common pattern in Pub/Sub applications because the create_topic method will raise an exception if the topic already exists. In production code, you might want to catch more specific exceptions, but for development and testing, this simple approach works well.

The topic acts as a buffer between your publishers and subscribers. When you publish a message, it gets stored in the topic until subscribers are ready to process it. This decoupling is what makes Pub/Sub so powerful for building resilient, scalable systems.

One of the most critical aspects of working with Pub/Sub is understanding message serialization. Pub/Sub messages must be sent as byte arrays, which means you need to convert your Python objects into a format that can be transmitted over the network and reconstructed by subscribers. It's important to note that the data field in every Pub/Sub message must be a non-empty bytestring (unless you provide at least one attribute); if you attempt to publish a message with an empty data field and no attributes, Pub/Sub will raise an error such as InvalidArgument. For example, json.dumps({}).encode() produces b'{}', which is valid because it is not empty, but if your serialization step accidentally produces an empty bytestring (like b''), your publish call will fail unless you include at least one attribute. Always ensure your message data is non-empty before publishing.

The most common serialization format is JSON because it's human-readable, widely supported, and language-agnostic. However, not all Python objects can be directly serialized to JSON. Here are the key serialization challenges you'll encounter:

JSON-Compatible Types:

• Strings, numbers, booleans, None
• Lists and dictionaries (if they contain JSON-compatible values)

Non-JSON-Compatible Types that require conversion:

• datetime objects → convert to ISO format strings using .isoformat()
• UUID objects → convert to strings using str(uuid_obj)
• set objects → convert to lists using list(set_obj)
• Custom objects → extract relevant attributes into dictionaries

Here's an example of handling these serialization challenges:

import json
import uuid
from datetime import datetime

# Original data with non-serializable types
signup_data = {
    "user_id": uuid.uuid4(),  # UUID object
    "timestamp": datetime.now(),  # datetime object
    "preferences": {"notifications", "marketing"}  # set object
}

# Convert to JSON-serializable format
signup_data["user_id"] = str(signup_data["user_id"])
signup_data["timestamp"] = signup_data["timestamp"].isoformat()
signup_data["preferences"] = list(signup_data["preferences"])

# Now it can be serialized
message_bytes = json.dumps(signup_data).encode()

The final .encode() step converts the JSON string into bytes, which is the format Pub/Sub requires. This two-step process (Python object → JSON string → bytes) is the standard pattern for message publishing.

Publishing messages involves creating structured event data and converting it into the proper format for Pub/Sub transmission. Let's build a comprehensive user signup event that demonstrates real-world complexity:

import json
import uuid
from datetime import datetime

def publish_signup():
    # Create event data with various Python types
    signup = {
        "user_id": uuid.uuid4(),
        "email": "user@example.com",
        "timestamp": datetime.now(),
        "plan": "free",
        "preferences": {"notifications", "marketing"}
    }
    
    # Handle serialization challenges
    signup["user_id"] = str(signup["user_id"])
    signup["timestamp"] = signup["timestamp"].isoformat()
    signup["preferences"] = list(signup["preferences"])
    
    # Publish the message
    return publisher.publish(tp, data=json.dumps(signup).encode()).result()

This function demonstrates several important concepts:

1. Real-world data complexity: The signup event includes different data types that you'd actually encounter in production applications.

2. Systematic serialization: Each non-JSON-compatible type is converted using the appropriate method. UUID objects become strings, datetime objects become ISO format strings, and sets become lists.

3. Two-step encoding: First json.dumps() creates a JSON string, then .encode() converts it to bytes.

4. Synchronous publishing: The .result() call blocks until the message is published and returns the message ID.

The order of serialization matters. You must convert all non-JSON-compatible types before calling json.dumps(), otherwise you'll get a TypeError about objects not being JSON serializable.

As you work with more complex events, you'll encounter additional serialization patterns. Here are some common scenarios and their solutions:

Working with nested objects:

user_profile = {
    "user_id": uuid.uuid4(),
    "created_at": datetime.now(),
    "metadata": {
        "last_login": datetime.now(),
        "session_id": uuid.uuid4()
    }
}

# Convert nested objects recursively
user_profile["user_id"] = str(user_profile["user_id"])
user_profile["created_at"] = user_profile["created_at"].isoformat()
user_profile["metadata"]["last_login"] = user_profile["metadata"]["last_login"].isoformat()
user_profile["metadata"]["session_id"] = str(user_profile["metadata"]["session_id"])

Working with lists of complex objects:

events = [
    {"event_id": uuid.uuid4(), "timestamp": datetime.now()},
    {"event_id": uuid.uuid4(), "timestamp": datetime.now()}
]

# Convert each item in the list
for event in events:
    event["event_id"] = str(event["event_id"])
    event["timestamp"] = event["timestamp"].isoformat()

Handling optional fields:

signup = {
    "user_id": uuid.uuid4(),
    "email": "user@example.com",
    "referral_code": uuid.uuid4() if has_referral else None
}

# Convert only if the field exists and isn't None
signup["user_id"] = str(signup["user_id"])
if signup["referral_code"]:
    signup["referral_code"] = str(signup["referral_code"])

These patterns will help you handle the variety of data structures you'll encounter in practice exercises.

Let's examine the complete solution to see how all these pieces work together in a cohesive application:

import os, json, uuid
from datetime import datetime
from google.cloud import pubsub_v1

os.environ.setdefault("PUBSUB_EMULATOR_HOST", "localhost:8681")
print("Pub/Sub emulator target →", os.environ["PUBSUB_EMULATOR_HOST"])

PROJECT, TOPIC = "demo-local", "user-signups"
publisher = pubsub_v1.PublisherClient()
tp = publisher.topic_path(PROJECT, TOPIC)
try:
    publisher.create_topic(name=tp)
except Exception:
    pass

def publish_signup():
    signup = {
        "user_id": uuid.uuid4(),
        "email": "user@example.com",
        "timestamp": datetime.now(),
        "plan": "free",
        "preferences": {"notifications", "marketing"}
    }
    
    # Convert non-serializable types to JSON-friendly formats
    signup["user_id"] = str(signup["user_id"])
    signup["timestamp"] = signup["timestamp"].isoformat()
    signup["preferences"] = list(signup["preferences"])
    
    return publisher.publish(tp, data=json.dumps(signup).encode()).result()

if __name__ == "__main__":
    print("published id →", publish_signup())

This complete implementation demonstrates the full workflow from environment setup through message publishing. Notice how the serialization handling is embedded directly in the publishing function, ensuring that every message is properly formatted before transmission.

The script begins by configuring the emulator connection and confirming the target. The publisher client and topic path creation happen at the module level for efficiency. The topic creation with exception handling ensures your topic exists before publishing.

When you run this complete script, you'll see output similar to:

Pub/Sub emulator target → localhost:8681
published id → 1234567890123456789

The message ID confirms that your properly serialized event data has been successfully stored in the topic, ready for subscribers to process.

Congratulations! You've successfully built your first event publisher using the Google Cloud Pub/Sub emulator with proper message serialization. You've learned how to configure your environment for local development, create publishers and topics, handle complex data serialization challenges, and publish well-structured messages.

The key concepts you've mastered include setting up the emulator connection, using the PublisherClient to interact with Pub/Sub, creating topics with proper error handling, and most importantly, converting Python data structures with various types (UUID, datetime, sets) into JSON-serializable formats before encoding to bytes for message publishing.

Your working publisher can now send complex user signup events with proper serialization handling. This foundation prepares you for the practice exercises where you'll encounter various data types and serialization challenges. You'll work with different event structures, handle edge cases in serialization, and explore message attributes while building on these core patterns.

In the upcoming practice exercises, you'll get hands-on experience with variations of this publisher pattern, including handling different data types, managing serialization errors, and working with more complex event structures. The next lesson will introduce you to the subscriber side of Pub/Sub, where you'll learn to create subscriptions that can receive and process the events you're now able to publish and properly serialize.