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

How to use dynamodb_table_exists method in localstack

Best Python code snippet using localstack_python

dynamodb_metadata_store.py

Source:dynamodb_metadata_store.py

...33    def close(self):34        pass35    def exists(self) -> bool:36        for table_name in [Settings.connections_table_name, Settings.variables_table_name, Settings.dag_deployments_table_name]:37            if not dynamodb_helper.dynamodb_table_exists(38                    ddb_client=self.client,39                    table_name=table_name,40            ):41                return False42        return True43    def _create_table_if_not_exists(44            self,45            table_name: str,46            primary_key: str,47            range_key: Optional[str] = None,48            read_capacity: int = 1,49            write_capacity: int = 1,50    ):51        if dynamodb_helper.dynamodb_table_exists(52                ddb_client=self.client,53                table_name=table_name,54        ):55            print(f'Table {table_name} exists. Skipping creation...')56        else:57            print(f'Creating table {table_name}...')58            kwargs = {'range_key': range_key} if range_key else {}59            dynamodb_helper.create_dynamodb_table(60                ddb_client=self.client,61                table_name=table_name,62                primary_key=primary_key,63                read_capacity_units=read_capacity,64                write_capacity_units=write_capacity,65                **kwargs...

dynamodb_helper.py

Source:dynamodb_helper.py

...45    while 'LastEvaluatedKey' in response:46        response = table.scan(ExclusiveStartKey=response['LastEvaluatedKey'])47        data.extend(response['Items'])48    return data49def dynamodb_table_exists(ddb_client, table_name: str):50    existing_tables = ddb_client.list_tables()['TableNames']51    return table_name in existing_tables52def create_dynamodb_table(53        ddb_client,54        table_name: str,55        primary_key: str,56        range_key: Union[str, None] = None,  # May have other types in the future57        read_capacity_units: int = 1,58        write_capacity_units: int = 1,59):60    key_schema = [61        {62            'AttributeName': primary_key,63            'KeyType': 'HASH'...

find_dangling_cloudwatch_alarms.py

Source:find_dangling_cloudwatch_alarms.py

...25#26# There's no point doing repeated, successive checks for the existence27# of a table/queue/function/whatever, so cache the results here.28@functools.cache29def dynamodb_table_exists(sess, *, table_name):30    """31    Returns True if this DynamoDB table exists, False otherwise.32    """33    client = sess.client("dynamodb")34    try:35        client.describe_table(TableName=table_name)36        return True37    except ClientError as err:  # pragma: no cover38        if err.response["Error"]["Code"] == "ResourceNotFoundException":39            return False40        else:41            raise42@functools.cache43def sqs_queue_exists(sess, *, queue_name):44    """45    Returns True if this SQS queue exists, False otherwise.46    """47    client = sess.client("sqs")48    try:49        client.get_queue_url(QueueName=queue_name)50        return True51    except ClientError as err:  # pragma: no cover52        if err.response["Error"]["Code"] == "AWS.SimpleQueueService.NonExistentQueue":53            return False54        else:55            raise56@functools.cache57def lambda_function_exists(sess, *, function_name):58    """59    Returns True if this Lambda function exists, False otherwise.60    """61    client = sess.client("lambda")62    try:63        client.get_function(FunctionName=function_name)64        return True65    except ClientError as err:  # pragma: no cover66        if err.response["Error"]["Code"] == "ResourceNotFoundException":67            return False68        else:69            raise70if __name__ == "__main__":71    sess = boto3.Session()72    for alarm in get_metric_alarm_descriptions(sess):73        dimensions = {dim["Name"]: dim["Value"] for dim in alarm["Dimensions"]}74        # Is this alarm based on a non-existent table?75        if alarm.get("Namespace") == "AWS/DynamoDB":76            table_name = dimensions["TableName"]77            if not dynamodb_table_exists(sess, table_name=table_name):78                print(79                    f"!!! Alarm {alarm['AlarmArn']} is based on non-existent table {table_name}"80                )81            continue82        # Is this alarm based on a non-existent queue?83        if alarm.get("Namespace") == "AWS/SQS":84            queue_name = dimensions["QueueName"]85            if not sqs_queue_exists(sess, queue_name=queue_name):86                print(87                    f"!!! Alarm {alarm['AlarmArn']} is based on non-existent SQS queue {queue_name}"88                )89            continue90        # Is this alarm based on a non-existent Lambda function?91        if alarm.get("Namespace") == "AWS/Lambda":...

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.