Google CloudEvents Python Library

0.14.0 · active · verified Wed Apr 15

The `google-events` library provides strongly-typed data payload schemas (using `proto-plus`) for various Google Cloud services that emit CloudEvents. It helps developers safely deserialize and interact with event data, ensuring type correctness and reducing runtime errors. The library is actively maintained by Google and sees frequent updates, primarily adding new event types and fields as Google Cloud services evolve.

Warnings

gotcha The `google-events` library primarily provides *data payload schemas* (Protobuf definitions) for Google CloudEvents. To work with full CloudEvent objects (which include context attributes like `id`, `source`, `type`, `time`, etc.), you typically need to use the separate `cloudevents` Python SDK (e.g., `pip install cloudevents`). `google-events` is designed to be used *with* a CloudEvents SDK, not as a standalone replacement.
Fix: Install `cloudevents` (`pip install cloudevents`) and use its `CloudEvent` class, then pass the `google-events` data object to `event.data` or deserialize `event.data` using a `google-events` schema.
gotcha Google CloudEvent data schemas are often versioned (e.g., `v1`). Always import the specific data type from its versioned module (e.g., `from google.events.cloud.storage.v1 import StorageObjectData`). Importing from a non-versioned parent module may not provide the correct or most up-to-date schema, and future updates to service schemas could introduce incompatible changes.
Fix: Explicitly use versioned imports, e.g., `from google.events.cloud.service.vX import EventData` where `vX` matches the event type version you expect.
gotcha When consuming raw event data (e.g., a dictionary or JSON string from an HTTP POST body or Pub/Sub message), you must deserialize it into the appropriate `google-events` proto-plus object using `.from_json()` or `.from_dict()`. Attempting to access attributes directly on the raw dictionary or parsing it into a generic object will bypass type safety and lead to runtime errors or incorrect data interpretation.
Fix: Always use `EventTypeData.from_json(json_string)` or `EventTypeData.from_dict(dictionary)` for deserialization before accessing event data fields.

Install

pip install google-events Install stable version

Imports

StorageObjectData
```
from google.events.cloud.storage.v1 import StorageObjectData
```
Event types are typically versioned; always import from the specific 'vX' module (e.g., v1).
MessagePublishedData
```
from google.events.cloud.pubsub.v1 import MessagePublishedData
```
Example for another common event type (Pub/Sub).

Quickstart

This quickstart demonstrates how to deserialize a raw JSON event data payload (e.g., from a Cloud Storage object finalization event) into a `StorageObjectData` object and access its strongly-typed properties.

import json
from google.events.cloud.storage.v1 import StorageObjectData

# Simulate a raw CloudEvent 'data' payload for a GCS object finalization
# This would typically come from an HTTP request body or Pub/Sub message.
raw_event_data = {
    "bucket": "my-test-bucket",
    "name": "my-folder/new-file.txt",
    "contentType": "text/plain",
    "size": "12345",
    "storageClass": "STANDARD",
    "timeCreated": "2023-10-27T10:00:00.000Z",
    "updated": "2023-10-27T10:00:00.000Z"
}

# Deserialize the raw JSON data into a strongly-typed StorageObjectData object
storage_object_data = StorageObjectData.from_json(json.dumps(raw_event_data))

# Access properties with type safety
print(f"Bucket: {storage_object_data.bucket}")
print(f"Object Name: {storage_object_data.name}")
print(f"Content Type: {storage_object_data.content_type}")
print(f"Size: {storage_object_data.size}")

# Example of a field that might not always be present (e.g., etag)
if storage_object_data.etag: # proto-plus fields return None or default for missing/empty
    print(f"ETag: {storage_object_data.etag}")
else:
    print("ETag not present in event data.")

view raw JSON →