def publish_to_pubsub()

in webhooks-with-cloud-run/MicroServices/Webhook/webhook.py [0:0]

• 14 lines of code
• 3 McCabe index (conditional complexity)

def publish_to_pubsub(msg):
    """
    Publishes the message to Cloud Pub/Sub
    """
    try:
        publisher = pubsub_v1.PublisherClient()
        topic_path = publisher.topic_path(
            os.environ.get("PROJECT_NAME"), os.environ.get("TOPIC_NAME")
        )

        # Pub/Sub data must be bytestring, attributes must be strings
        future = publisher.publish(topic_path, data=msg)

        exception = future.exception()
        if exception:
            raise Exception(exception)

        print(f"Published message: {future.result()}")

    except Exception as e:
        # Log any exceptions to stackdriver
        entry = dict(severity="WARNING", message=e)
        print(entry)