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

How to use _queue_exists method in localstack

Best Python code snippet using localstack_python

pgjsonb_queue.py

Source:pgjsonb_queue.py

...128    Provide the number of items in a queue129    """130    items = _list_items(queue)131    return len(items)132def _queue_exists(queue):133    """134    Does this queue exist135    :param queue: Name of the queue136    :type str137    :return: True if this queue exists and138    False otherwise139    :rtype bool140    """141    return queue in list_queues()142def handle_queue_creation(queue):143    if not _queue_exists(queue):144        with _conn(commit=True) as cur:145            log.debug("Queue %s does not exist. Creating", queue)146            _create_table(cur, queue)147    else:148        log.debug("Queue %s already exists.", queue)149def insert(queue, items):150    """151    Add an item or items to a queue152    """153    handle_queue_creation(queue)154    with _conn(commit=True) as cur:155        if isinstance(items, dict):156            items = salt.utils.json.dumps(items)157            cmd = str("""INSERT INTO {0}(data) VALUES('{1}')""").format(...

sqs_listener.py

Source:sqs_listener.py

...116            use_ssl=True,117            verify=verify118        )119        self._client = client120        if (not self._queue_exists()): 121            raise ValueError(f'The Queue {self._options.queue_name} does not exist')122        123        return client124    125    def _queue_exists(self):126        ''' Gets a list of available queues to verify the queue were looking at exists. '''127        queues = self._client.list_queues(QueueNamePrefix=self._options.queue_name)128        if 'QueueUrls' in queues:129            for q in queues['QueueUrls']:130                name = q.split('/')[-1]131                if name == self._options.queue_name:132                    return True133                134    def _get_queue_url(self): 135        ''' Uses the boto3 client to find and set the QueueUrl '''136        # Is the queue url already set?137        if (self._options.queue_url):138            return139            ...

queues.py

Source:queues.py

2from bob.common.aws import get_boto3_resource3from bob.worker.aws_helpers import error_code_equals4_task_queue_name = 'bob-task'5#_task_queue_name = 'bob-task-test'6def _queue_exists(queue_name, sqs):7    try:8        queue = sqs.get_queue_by_name(QueueName=queue_name)9        print(queue.url)10        return True11    except ClientError as err:12        if error_code_equals(err, 'AWS.SimpleQueueService.NonExistentQueue'):13            return False14        raise err15def create_task_queue(sqs=get_boto3_resource('sqs')):16    if _queue_exists(_task_queue_name, sqs=sqs):17        return18    sqs.create_queue(QueueName=_task_queue_name,19                     Attributes={'VisibilityTimeout': '60',20                                 'ReceiveMessageWaitTimeSeconds': '15'})21def _create_task_cancel_queue(sqs=get_boto3_resource('sqs')):22    if _queue_exists(_task_queue_name, sqs=sqs):23        return24    sqs.create_queue(QueueName=_task_queue_name,25                     Attributes={'VisibilityTimeout': '60',26                                 'ReceiveMessageWaitTimeSeconds': '15'})27def enqueue_task(task, sqs=get_boto3_resource('sqs')):28    queue = sqs.get_queue_by_name(QueueName=_task_queue_name)29    queue.send_message(MessageBody=str(task))30def get_task_queue(sqs=get_boto3_resource('sqs')):...

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.