"""SQS url parsing module""" import re def parse_env(section): """Logic to extract environment from a section of SQS queue name. Args: section(str): Section of an SQS url like 'prod-delivery17', 'e0000001' Returns: env(str|None): returns environment if sections contains it; else returns None """ env = None if 'prod' in section: env = 'prod' elif 'dev' in section: env = 'dev' elif 'qa' in section: env = 'qa' return env def parse_priority(section): """Logic to extract priority from a section of SQS queue name. Args: section(str): Section of an SQS url like 'prod-delivery17', 'e0000001' Returns: priority(int): returns the priority if sections contains it; else returns None """ return int(section.strip('ed').lstrip('0')) def parse_encoder_id(section): """Logic to extract encoder ID from a section of SQS queue name. Args: section(str): Section of an SQS url like 'prod-delivery17', 'e0000001' Returns: encoder_id(int|None): returns the encoder ID if sections contains it; else returns None """ envs = ['dev', 'prod', 'qa'] encoder_id = None if any(s in section for s in envs): encoder_id = int(re.sub('[^0-9]', '', section)) return encoder_id def parse_queue_name(queue_name): """Logic to extract tags from the SQS queue name. Args: section(str): Section of an SQS url like 'prod-delivery17', 'e0000001' Returns: tags(tuple): tags from the queue_name in the order ['env', 'dpm_priority', 'store_priority'] """ sections = queue_name.split('_') if len(sections) != 3: raise ValueError('Badly formatted queue_name: {}'.format(queue_name)) tags = (parse_env(sections[0]), parse_encoder_id(sections[0]), parse_priority(sections[1]), parse_priority(sections[2])) return tags