Best Python code snippet using localstack_python
db.py
Source:db.py  
1import boto32import botocore3import json4import config5import res.utils as utils6import res.glob  as glob7# =======================================================================================================================8#9#  Supported services   : RDS, DynamoDB, ElastiCache, Neptune, Amazon Redshift10#  Unsupported services : None11#12# =======================================================================================================================13#  ------------------------------------------------------------------------14#15#    RDS 16#17#  ------------------------------------------------------------------------18def get_rds_inventory(oId):19    """20        Returns RDS inventory21        :param oId: ownerId (AWS account)22        :type oId: string23        :return: RDS inventory24        :rtype: json25        ..note:: http://boto3.readthedocs.io/en/latest/reference/services/rds.html26    """27    return glob.get_inventory(28        ownerId = oId,29        aws_service = "rds", 30        aws_region = "all", 31        function_name = "describe_db_instances", 32        key_get = "DBInstances",33        pagination = True34    )35#  ------------------------------------------------------------------------36#37#    DynamoDB 38#39#  ------------------------------------------------------------------------40def get_dynamodb_inventory(oId):41    """42        Returns dynamoDB inventory43        :param oId: ownerId (AWS account)44        :type oId: string45        :return: dynamoDB inventory46        :rtype: json47        ..note:: http://boto3.readthedocs.io/en/latest/reference/services/dynamodb.html48    """49    return glob.get_inventory(50        ownerId = oId,51        aws_service = "dynamodb", 52        aws_region = "all", 53        function_name = "list_tables", 54        key_get = "TableNames",55        detail_function = "describe_table", 56        join_key = "TableName", 57        detail_join_key = "TableName", 58        detail_get_key = "Table",59        pagination = True60    )61#  ------------------------------------------------------------------------62#63#    Neptune 64#65#  ------------------------------------------------------------------------66def get_neptune_inventory(oId):67    """68        Returns neptune inventory (instances & clusters). Instances are listed in RDS inventory.69        :param oId: ownerId (AWS account)70        :type oId: string71        :return: neptune inventory72        :rtype: json73        ..note:: http://boto3.readthedocs.io/en/latest/reference/services/neptune.html74    """75    neptune_inventory = {}76    neptune_inventory['clusters'] = glob.get_inventory(77        ownerId = oId,78        aws_service = "neptune", 79        aws_region = "all", 80        function_name = "describe_db_clusters", 81        key_get = "DBClusters"82    )83    return neptune_inventory84   85#  ------------------------------------------------------------------------86#87#    ElastiCache88#89#  ------------------------------------------------------------------------90def get_elasticache_inventory(oId):91    """92        Returns elasticache inventory (instances & clusters). Instances are listed in RDS inventory.93        :param oId: ownerId (AWS account)94        :type oId: string95        :return: elasticache inventory96        :rtype: json97        ..note:: http://boto3.readthedocs.io/en/latest/reference/services/elasticache.html98    """99    elasticache_inventory = {}100    elasticache_inventory['cache-clusters'] = glob.get_inventory(101        ownerId = oId,102        aws_service = "elasticache", 103        aws_region = "all", 104        function_name = "describe_cache_clusters", 105        key_get = "CacheClusters",106        pagination = True107    )108    elasticache_inventory['reserved-cache-nodes'] = glob.get_inventory(109        ownerId = oId,110        aws_service = "elasticache", 111        aws_region = "all", 112        function_name = "describe_reserved_cache_nodes", 113        key_get = "ReservedCacheNodes",114        pagination = True115    )116    return elasticache_inventory117   118#  ------------------------------------------------------------------------119#120#    Redshift121#122#  ------------------------------------------------------------------------123def get_redshift_inventory(oId):124    """125        Returns redshift inventory (instances & clusters). Instances are listed in RDS inventory.126        :param oId: ownerId (AWS account)127        :type oId: string128        :return: redshift inventory129        :rtype: json130        ..note:: http://boto3.readthedocs.io/en/latest/reference/services/redshift.html131    """132    redshift_inventory = {}133    redshift_inventory['clusters'] = glob.get_inventory(134        ownerId = oId,135        aws_service = "redshift", 136        aws_region = "all", 137        function_name = "describe_clusters", 138        key_get = "Clusters",139        pagination = True140    )141    redshift_inventory['reserved-nodes'] = glob.get_inventory(142        ownerId = oId,143        aws_service = "redshift", 144        aws_region = "all", 145        function_name = "describe_reserved_nodes", 146        key_get = "ReservedNodes",147        pagination = True148    )149    return redshift_inventory150#151# Hey, doc: we're in a module!152#153if (__name__ == '__main__'):...export-reserved.py
Source:export-reserved.py  
...29                wrt_result = '%s,%s,%s,%s,%s,%s,%s\n' % (resvere_ins.get('ProductDescription'), resvere_ins.get('DBInstanceClass'), resvere_ins.get('DBInstanceCount'), expired_time, region_name, profile, resvere_ins.get('MultiAZ'))30                f.write(wrt_result)31def redshiftreserve(session, report_file, region_name, profile):32    redshift = session.client('redshift')33    response = redshift.describe_reserved_nodes()34    if not response.get('ReservedDBInstances'):35        return36    with file(report_file, 'a') as f:37        f.seek(0,2)38        for resvere_ins in response.get('ReservedDBInstances'):39            if resvere_ins.get('State') == 'active':40                duration = resvere_ins.get('Duration')41                time_delta = datetime.timedelta(seconds=duration)42                expired_time = resvere_ins.get('StartTime') + time_delta43                wrt_result = '%s,%s,%s,%s,%s,%s\n' % ('Redshift', resvere_ins.get('NodeType'), resvere_ins.get('NodeCount'), expired_time, region_name, profile)44                f.write(wrt_result)45def cachereserve(session, report_file, region_name, profile):46    elasticache = session.client('elasticache')47    response = elasticache.describe_reserved_cache_nodes()...finder.py
Source:finder.py  
...33    print ([ec2ReservedID['ReservedInstancesId'] for ec2ReservedID in reservedEC2Instances['ReservedInstances']])34        35    reservedRdsInstances = rds.describe_reserved_db_instances()36    print ("RDS reserved instances are:"[rdsReservedID['ReservedDBInstanceId'] for rdsReservedID in reservedRdsInstances['ReservedDBInstances']])37    redshiftReservedNodes = redshift.describe_reserved_nodes()38    print ("Redshift reserved nodes are:"[redshiftReservedID['ReservedNodeId'] for redshiftReservedID in redshiftReservedNodes['ReservedNodes']])39    esReserved = es.describe_reserved_cache_nodes()...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.
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Get 100 minutes of automation test minutes FREE!!
