google_cloudevents_storage_event_parsing_quickstart.py

1from google.events.cloud import storage_v1
2from cloudevents.http import from_http
3
4def hello_gcs(request):
5    """Responds to a Cloud Storage event.
6    Args:
7        request (flask.Request): The request object.
8    """
9    # Create a CloudEvent from the HTTP request
10    event = from_http(request.headers, request.get_data())
11
12    # Check the event type
13    if event["type"] == "google.cloud.storage.object.v1.finalized":
14        # Wrap the event data in a typed StorageObjectData object
15        data = storage_v1.StorageObjectData.from_json(event.data)
16
17        print(f"Event ID: {event['id']}")
18        print(f"Event Type: {event['type']}")
19        print(f"Bucket: {data.bucket}")
20        print(f"File: {data.name}")
21        print(f"Metageneration: {data.metageneration}")
22        print(f"Created: {data.time_created}")
23
24    return "OK"