Skip to content

Instantly share code, notes, and snippets.

Created November 22, 2022 12:43
  • Star 0 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
Star You must be signed in to star a gist
What would you like to do?
Sample Events Generator for Google Cloud Pub/Sub
"""Publishes multiple random messages to a Pub/Sub topic."""
from import pubsub_v1
from datetime import datetime
import random
import json
project_id = "stddevco"
topic_id = "events-demo"
browsers = ['Chrome','Opera','Firefox','Safari']
OSs = ['Windows','Linux','OSX','iOS','Android']
event_types = ['visit','click','add_to_cart','buy_cart','abandoned_cart']
publisher = pubsub_v1.PublisherClient()
# The `topic_path` method creates a fully qualified identifier
# in the form `projects/{project_id}/topics/{topic_id}`
topic_path = publisher.topic_path(project_id, topic_id)
while True:
product_id = random.randrange(10000, 11000, 1)
event_type = random.choice(event_types)
event = {
'timestamp': datetime.utcnow().isoformat(),
'event': event_type,
'product_id': product_id if event_type in ['visit','add_to_cart'] else None,
'url': f'product/{product_id}?foo=bar' if event_type in ['visit','add_to_cart'] else None,
'browser': random.choice(browsers),
'OS': random.choice(OSs),
'cart_id': random.randrange(1_000_000,1_100_000) if event_type.endswith('cart') else None
data_str = json.dumps(event)
data = data_str.encode("utf-8")
# When you publish a message, the client returns a future.
future = publisher.publish(topic_path, data)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment