def stream_event()

in app/helpers/eventing/helpers.py [0:0]

• 9 lines of code
• 1 McCabe index (conditional complexity)

def stream_event(topic_name, event_type, event_context):
    """
    Helper function for publishing an event.

    Parameters:
       topic_name (str): The name of the Cloud Pub/Sub topic.
       event_type (str): The type of the event.
       event_context: The context of the event.

    Output:
       None.
    """

    topic_path = publisher.topic_path(GCP_PROJECT, topic_name)
    request = {
        'event_type': event_type,
        'created_time': str(int(time.time())),
        'event_context': event_context
    }
    data = json.dumps(request).encode()
    publisher.publish(topic_path, data)