KaneAI - World’s First E2E Software Testing Agent.Signup for Private Beta Join TestMu Conference

How to use wait_time_seconds method in localstack

Best Python code snippet using localstack_python

decorators.py

Source:decorators.py

1import asyncio2import typing as tp3from functools import wraps4from time import sleep5from toolset.typing_helpers import ASYNC_FUNC, FUNC_RESULT, TFunc6def retry(  # noqa: WPS231 Found too high function cognitive complexity7    *exceptions: tp.Type[Exception],8    attempts: int = 3,9    wait_time_seconds: float = 0.5,10    backoff: int = 2,11) -> tp.Callable[[TFunc], TFunc]:12    """Try to call a func `attempts` times with `wait_time_seconds` breaks multiplied each time by `backoff`."""13    def _retry(func: TFunc) -> TFunc:14        @wraps(func)15        def _inner(*args, **kwargs) -> FUNC_RESULT:16            exception: Exception = Exception()17            wait = wait_time_seconds18            for _ in range(attempts):  # noqa: WPS122 unused variables definition19                try:20                    res: FUNC_RESULT = func(*args, **kwargs)21                except exceptions as exc:22                    exception = exc23                    sleep(wait)24                    wait *= backoff25                    continue26                break27            else:28                # raise catch exception29                raise exception30            return res31        return tp.cast(TFunc, _inner)32    return _retry33def aio_retry(  # noqa: WPS231 Found too high function cognitive complexity34    *exceptions: tp.Type[Exception],35    attempts: int = 3,36    wait_time_seconds: float = 0.5,37    backoff: int = 2,38):39    """Try to call a func `attempts` times with `wait_time_seconds` breaks multiplied each time by `backoff`."""40    def _retry(func: ASYNC_FUNC) -> ASYNC_FUNC:41        @wraps(func)42        async def _inner(*args, **kwargs) -> FUNC_RESULT:43            exception: Exception = Exception()44            wait = wait_time_seconds45            for _ in range(attempts):  # noqa: WPS122 unused variables definition46                try:47                    res = await func(*args, **kwargs)48                except exceptions as exc:49                    exception = exc50                    await asyncio.sleep(wait)51                    wait *= backoff52                    continue53                return res54            # raise catch exception55            raise exception56        return tp.cast(ASYNC_FUNC, _inner)...

qprocessor.py

Source:qprocessor.py

1ï»¿# Simple Python SQS consumer2# Conor Murphy34import sys56# must install boto3 to use this (pip3 install boto3)7import boto389# get QueueUrl from command line10if len(sys.argv) < 2:11    print('usage: python3 qprocessor.py https://queue_url_here/ out_file_path')12    sys.exit(1)13queue_url=sys.argv[1]14out_file_path=sys.argv[2]1516# client object for SQS17sqs = boto3.client('sqs')1819wait_time_seconds = 12021def print_to_file(line):22    out_file=open(out_file_path,'a')23    print(line, file=out_file)24    out_file.close()2526print_to_file('START')2728while True:29    # receive30    response = sqs.receive_message(QueueUrl=queue_url, WaitTimeSeconds=wait_time_seconds, MaxNumberOfMessages=1)31    wait_time_seconds = 203233    if 'Messages' not in response:34        print('no messages hit Ctrl + Z to exit queue')35        continue3637    # loop over all messages38    for message in response['Messages']:39        40        # do processing work here (example just prints)41        print("received: " + message['Body'])42        print_to_file(message['Body'])43        44        45        # delete once processed46        sqs.delete_message(QueueUrl=queue_url, ReceiptHandle=message['ReceiptHandle'])47        
...

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.