How to access Google Pub/Sub Custom Attributes

Question

I have a set of Cloud Functions that send Pub/Sub messages to other functions in my pipeline. One of these functions is an email function. Ideally, I'd like to use Jinja2 templates in this email function that get filled in based on the custom attributes of my Pub/Sub messages. My code that sends the message is as follows:

def publish_detailed_message(project: str,
                             topic: str,
                             message: str,
                             **kwargs):
    """publish_message posts a message to a PubSub topic.

    Args:
        project (str): the project that the topic is located
        topic (str): the topic to publish the message
        message (str): the message to be sent
    """
    publisher = pubsub_v1.PublisherClient(
        publisher_options=pubsub_v1.types.PublisherOptions(
            enable_message_ordering=True,
        )
    )
    topic_path = publisher.topic_path(project, topic)

    attributes = {}
    data_str = f"{message}"
    data = data_str.encode('utf-8')
    if kwargs:
        for k, v in kwargs.items():
            attributes[str(k)] = str(v)
        future = publisher.publish(topic_path, data, **attributes)
    else:
        future = publisher.publish(topic_path, data)
    print(future.result())
    print(f"Published messages to {topic_path}")

I've been able to successfully access these attributes when including the following in my def main():

        ATTR = {'Team': 'Arch', 'Data': 'Production'}
        publish_message(PROJECT,
                        TOPIC,
                        MESSAGE,
                        **ATTR)

This has allowed me to print the attributes, as well as recognize them in the Pub/Sub message itself. What I'd like to do is grab these attributes with my email function that is triggered by the Pub/Sub message and then pass these attributes into my Jinja2 template. I have successfully been able to log the attributes with the following generic cloud function.

    """The entrypoint for the work."""
    name = base64.b64decode(event['data']).decode('utf-8')
    print(
        """This Function was triggered by messageId {} published at {} to {}
    """.format(
            context.event_id, context.timestamp, context.resource["name"]
        )
    )
 
    logger.info(f'attributes - {event["attributes"]}')
    logger.info(f'content = {event["data"]}')
    logger.info(f'name - {name}')

which gives me the following log:

2022-07-27T20:30:28.705193Z
send-emailsu5jrky5cj5q [I 220727 20:30:28 main:32] attributes - {'Data': 'Production', 'Team': 'Arch'} 

However, when I try to build a dictionary using the event['attributes'] I can't get any info. I am using the publish_message function shown above to build dictionaries using the attributes but the following

    if event.attributes:
        attributes = {}
        for key in event.attributes:
            value = event.attributes.get(key)
            attributes[key] = value
        u.send_template_email('test.html',
                              TO,
                              'test',
                              **attributes)
    else:
        u.send_template_email('test.html',
                              TO,
                              'test')

tells me that the term 'attributes' is undefined

jinja2.exceptions.UndefinedError: 'attributes' is undefined 

I also tried accessing the attributes in a similar fashion to accessing the data,

event['attributes'].items()

but received errors for that as well. I feel as though this must be possible but am not able to find any documentation helping me through this in Python.

Ideally others have had this issue and may be able to point me in the right direction.

Answer 1

I looked some of my old code...

In a function that is triggered by a message from a pubsub topic, I had something like this:

# parse event attributes
if "attributes" in event and event["attributes"]:
    attr_dic = event["attributes"]    
    # handle attribute_1
    if "attribute_1" in attr_dic and attr_dic["attribute_1"]:
        # do something useful

and so on including handling incorrect cases and exceptions...

In a function, where the message is published to the pubsub topic, I had something like this:

PS.publish(
    PS_TOPIC, message_text_value.encode(),
    attribute_1=attribute_1_value,
    attribute_2=attribute_2_value)

Bear in mind that the names of the attributes and attribute values have some restrictions - see Use Attributes

Here is an example in the documentation: Publish with custom attributes

Here is how it is implemented: GitHub - python-pubsub/google/cloud/pubsub_v1/publisher/client.py