in helpers/bq-remote-functions/get-policy-tags/main.py [0:0]
def get_policy_tag_display_names(policy_tags, cache):
"""Looks up the display names of policy tags using Data Catalog."""
datacatalog_client = datacatalog_v1.PolicyTagManagerClient()
tag_ids = list(policy_tags.values())
display_names = {}
for tag_id in tag_ids:
if tag_id:
try:
# 1) get from datastore
cached_policy_tag_display_name = cache.get(tag_id)
if cached_policy_tag_display_name:
display_names[tag_id] = cached_policy_tag_display_name
else:
# API call
tag = datacatalog_client.get_policy_tag(name=tag_id)
display_names[tag_id] = tag.display_name
cache.add(tag_id, tag.display_name)
except Exception as e:
display_names[tag_id] = f'Failed to retrieve policy tag display name for {tag_id}. Exception: {e}'
return display_names