"""Helper functions for interacting with the cache.""" from typing import List from ddtrace import tracer from permissions.connectors import redis def get_resources_key(profile_type, profile_id, resource_type=None): """Get the key for indexing profile resource entries in the cache.""" res_type = resource_type or '*' return f'resources_for_{profile_type}_{profile_id}_{res_type}' @tracer.wrap('bust_by_identity', service='redis') def bust_by_identity( identity_uuids: List[str], resource_types: List[str], ) -> None: """Delete keys from redis based on identity_uuids and resource types. identity_uuids: List of identity uuids. resource_types: List of resource types. """ keys = [] for resource_type in resource_types: for identity_uuid in identity_uuids: keys.append(get_resources_key('identity', identity_uuid, resource_type)) redis.delete_keys(keys)