KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use get_tag_values method in localstack

Best Python code snippet using localstack_python

ofx_sanitize.py

Source:ofx_sanitize.py

...68            # Do generic replacements69            out.write(inner)70        else:71            numbers = []72            units = get_tag_values(inner, 'UNITS')73            unitprice = get_tag_values(inner, 'UNITPRICE')74            if len(units) == 1 and len(unitprice) == 1:75                numbers.append(D(units[0]) * D(unitprice[0]))76            for inner_tag in ['FEES', 'COMMISSION']:77                for value in get_tag_values(inner, inner_tag):78                    if value:79                        numbers.append(D(value))80            if numbers:81                total_number = sum(numbers, D('0'))82                inner = replace_tag_value(inner, 'TOTAL', lambda _: str(-total_number))83            out.write('<' + tag + '>' + inner + '</' + tag + '>')84    return out.getvalue()85def replace_tag_contents(contents: str, tag: str,86                         replacement: Callable[[str], str]) -> str:87    return re.sub(88        r'(<' + tag + '>)((?:[^<]|<(?!/' + tag + r'>))*)(</' + tag + r'>)',89        lambda m: m.group(1) + replacement(m.group(2)) + m.group(3),90        contents,91        flags=re.IGNORECASE)92def get_tag_values(contents: str, tag: str) -> List[str]:93    values = []  # type: List[str]94    def replacer(x):95        values.append(x)96        return x97    replace_tag_value(contents, tag, replacer)98    return values99def pretty_print(contents: str) -> str:100    out = io.StringIO()101    def recurse(level: int, s: str):102        indent = '  ' * level103        for tag, inner in get_balanced_tags(s):104            inner = inner.lstrip()105            if tag is None:106                for line in inner.replace('<', '\n<').lstrip().split('\n'):107                    out.write(indent + line + '\n')108                continue109            out.write(indent + '<' + tag + '>\n')110            recurse(level + 1, inner)111            out.write(indent + '</' + tag + '>\n')112    recurse(0, contents)113    return out.getvalue()114def get_strings(contents: str) -> List[str]:115    return [116        x.strip() for x in re.split('\n|</?[a-zA-Z][a-zA-Z0-9]*>', contents)117    ]118class MemoizedReplacer(object):119    def __init__(self, replacer: Callable[[str], str]) -> None:120        self.replacer = replacer121        self.replacements = dict()  # type: Dict[str, str]122    def __call__(self, x: str) -> str:123        return self.replacements.setdefault(x, self.replacer(x))124def sanitize_ofx(input_path: str, output_path: str, account_id: str,125                 broker_id: str, org: str, retained_tickers: FrozenSet[str]):126    with open(input_path, 'r') as f:127        contents = f.read()128    orig_strings = set(get_strings(contents))129    contents = re.sub(130        r'^([A-Z]*FILEUID:)(.*)$',131        lambda m: m.group(1) + make_random_hex_replacement(m.group(2)),132        contents,133        flags=re.MULTILINE)134    contents = replace_tag_value(contents, 'SESSCOOKIE',135                                 make_random_replacement)136    contents = replace_tag_value(contents, 'CLTCOOKIE',137                                 make_random_replacement)138    contents = replace_tag_value(contents, 'TRNUID',139                                 make_random_hex_replacement)140    contents = replace_tag_value(contents, 'ORG', lambda _: org)141    contents = replace_tag_value(contents, 'ACCTID', lambda _: account_id)142    contents = replace_tag_value(contents, 'BROKERID', lambda _: broker_id)143    contents = replace_tag_value(contents, 'FITID', make_random_replacement)144    contents = replace_tag_value(145        contents, 'UNIQUEID',146        MemoizedReplacer(lambda x: make_random_replacement(x).upper()))147    contents = replace_tag_value(contents, 'FIID',148                                 make_random_digits_replacement)149    contents = replace_tag_value(contents, 'FID',150                                 make_random_digits_replacement)151    contents = replace_tag_value(152        contents, 'TICKER',153        MemoizedReplacer(lambda x: make_random_ticker(x, retained_tickers)))154    contents = replace_tag_value(contents, 'SECNAME', lambda _: 'Security')155    allowed_strings = set([156        'Y',157        'N',158        'CUSIP',159        'BUY',160        'DIV',161        'INTEREST',162        'LONG',163        'SHORT',164        'SELL',165        'SUCCESS',166        'DEBIT',167        'CASH',168        'MATCH',169        'PRETAX',170        'AFTERTAX',171        'PERCENT',172        '0',173        'DOLLAR',174        'OTHER',175        'OPENEND',176        'CLOSEEND',177        'OLDFILEUID:NONE',178        'Price as of date based on closing price',179    ])180    # Allow dates181    for tag in [182            'DTASOF', 'DTPRICEASOF', 'DTTRADE', 'DTSTART', 'DTEND', 'DTSERVER',183            'DTPOSTED', 'DTPROFUP',184    ]:185        allowed_strings.update(get_tag_values(contents, tag))186    # Allow currencies187    allowed_strings.update(get_tag_values(contents, 'CURDEF'))188    for tag in ['BALTYPE', 'LANGUAGE', 'SEVERITY', 'MESSAGE']:189        allowed_strings.update(get_tag_values(contents, tag))190    # Allow BAL DESC and NAME values191    for tag, inner in get_balanced_tags(contents, 'BAL'):192        if tag is None: continue193        allowed_strings.update(get_tag_values(inner, 'DESC'))194        allowed_strings.update(get_tag_values(inner, 'NAME'))195    contents = replace_transactions(contents)196    for tag in [197            'MKTVAL',198            'BUYPOWER',199            'VALUE',200            'AVAILCASH',201            'MARGINBALANCE',202            'SHORTBALANCE',203    ]:204        contents = replace_tag_value(contents, tag, replace_number)205    retained_strings = orig_strings & set(206        get_strings(contents)) - allowed_strings207    allowed_strings_pattern = '^(CHARSET|COMPRESSION|DATA|ENCODING|OFXHEADER|SECURITY|VERSION):.*|^$'208    retained_strings = {...

delete_inactive_ebs.py

Source:delete_inactive_ebs.py

...11log = logging.getLogger(__name__)12log.setLevel(logging.INFO)13ATTACH_DATE_TAG_NAME = 'AttachDate'14INACTIVE_DAY_THRESHOLD = 515def get_tag_values(tag_name):16    """17    Call Resource Group Tagging API to get all values of a certain tag18    :param tag_name: Tag name19    :return: List of all values of the tag20    """21    client = boto3.client('resourcegroupstaggingapi')22    tag_values = []23    try:24        response = client.get_tag_values(25            Key=tag_name26        )27        tag_values += response['TagValues']28        while response and response.get('PaginationToken', ''):29            response = client.get_tag_values(30                PaginationToken=response['PaginationToken'],31                Key=tag_name32            )33            tag_values += response['TagValues']34        return tag_values35    except Exception as e:36        log.error(f'Failed to get tag values for {tag_name}.')37        log.error(e)38        return []39def delete_volumes(inactive_dates):40    """41    Find and delete all EBS volumes that have AttachDate tag value in inactive_dates42    :param inactive_dates: List of String that considered as inactive dates, for example, ["2021-04-27", "2021-04-26"]43    :return: Number of EBS volumes that are deleted successfully, number of EBS volumes that are not deleted.44    """45    success = 046    failure = 047    ec2_client = boto3.resource('ec2')48    response = ec2_client.volumes.filter(Filters=[49        {50            'Name': f'tag:{ATTACH_DATE_TAG_NAME}',51            'Values': inactive_dates52        },53        {54            'Name': 'status',55            'Values': ['available']56        },57    ])58    log.info(f'Deleting inactive EBS volumes that haven\'t been used for {INACTIVE_DAY_THRESHOLD} days...')59    for volume in response:60        if volume.attachments:61            # Double check if volume has an attachment in case the volume is being used.62            continue63        try:64            log.info(f'Deleting volume {volume.volume_id}')65            volume.delete()66            success += 167        except Exception as e:68            log.error(f'Failed to delete volume {volume.volume_id}.')69            log.error(e)70            failure += 171    return success, failure72def main():73    tag_values = get_tag_values(ATTACH_DATE_TAG_NAME)74    date_today = datetime.today().date()75    # Exclude dates of last {INACTIVE_DAY_THRESHOLD} days from tag values, the rest values are considered as inactive dates.76    dates_to_exclude = []77    for i in range(INACTIVE_DAY_THRESHOLD):78        dates_to_exclude.append(str(date_today - timedelta(i)))79    inactive_dates = set(tag_values).difference(dates_to_exclude)80    if inactive_dates:81        success, failure = delete_volumes(list(inactive_dates))82        log.info(f'{success} EBS volumes are deleted successfully.')83        if failure:84            log.info(f'{failure} EBS volumes are not deleted due to errors. See logs for more information.')85if __name__ == "__main__":...

resource_groups_tagging_api.py

Source:resource_groups_tagging_api.py

2from botocore.exceptions import ClientError34TAG_KEY_ENV_NAME = "EnvironmentName"56def get_tag_values(rgta_client, tag_key=TAG_KEY_ENV_NAME):7    try:8        paginator = rgta_client.get_paginator('get_tag_values')910        # Create a PageIterator from the Paginator object11        page_iterator = paginator.paginate(Key=tag_key)1213        # Return the list of values for a given tag14        tag_values = []15        for page in page_iterator:16            tag_values.extend(page["TagValues"])17        return tag_values18    except ClientError as e:19        raise Exception(f"boto3 client error in get_tag_values: {e}")20    except Exception as e:
...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.