Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use visibility_timeout method in localstack

Best Python code snippet using localstack_python

queue.py

Source:queue.py

1# -*- coding: utf-8 -*-2"""Amazon SQS queue implementation."""3from __future__ import absolute_import, unicode_literals4from vine import transform5from .message import AsyncMessage6_all__ = ['AsyncQueue']7def list_first(rs):8    """Get the first item in a list, or None if list empty."""9    return rs[0] if len(rs) == 1 else None10class AsyncQueue():11    """Async SQS Queue."""12    def __init__(self, connection=None, url=None, message_class=AsyncMessage):13        self.connection = connection14        self.url = url15        self.message_class = message_class16        self.visibility_timeout = None17    def _NA(self, *args, **kwargs):18        raise NotImplementedError()19    count_slow = dump = save_to_file = save_to_filename = save = \20        save_to_s3 = load_from_s3 = load_from_file = load_from_filename = \21        load = clear = _NA22    def get_attributes(self, attributes='All', callback=None):23        return self.connection.get_queue_attributes(24            self, attributes, callback,25        )26    def set_attribute(self, attribute, value, callback=None):27        return self.connection.set_queue_attribute(28            self, attribute, value, callback,29        )30    def get_timeout(self, callback=None, _attr='VisibilityTimeout'):31        return self.get_attributes(32            _attr, transform(33                self._coerce_field_value, callback, _attr, int,34            ),35        )36    def _coerce_field_value(self, key, type, response):37        return type(response[key])38    def set_timeout(self, visibility_timeout, callback=None):39        return self.set_attribute(40            'VisibilityTimeout', visibility_timeout,41            transform(42                self._on_timeout_set, callback,43            )44        )45    def _on_timeout_set(self, visibility_timeout):46        if visibility_timeout:47            self.visibility_timeout = visibility_timeout48        return self.visibility_timeout49    def add_permission(self, label, aws_account_id, action_name,50                       callback=None):51        return self.connection.add_permission(52            self, label, aws_account_id, action_name, callback,53        )54    def remove_permission(self, label, callback=None):55        return self.connection.remove_permission(self, label, callback)56    def read(self, visibility_timeout=None, wait_time_seconds=None,57             callback=None):58        return self.get_messages(59            1, visibility_timeout,60            wait_time_seconds=wait_time_seconds,61            callback=transform(list_first, callback),62        )63    def write(self, message, delay_seconds=None, callback=None):64        return self.connection.send_message(65            self, message.get_body_encoded(), delay_seconds,66            callback=transform(self._on_message_sent, callback, message),67        )68    def write_batch(self, messages, callback=None):69        return self.connection.send_message_batch(70            self, messages, callback=callback,71        )72    def _on_message_sent(self, orig_message, new_message):73        orig_message.id = new_message.id74        orig_message.md5 = new_message.md575        return new_message76    def get_messages(self, num_messages=1, visibility_timeout=None,77                     attributes=None, wait_time_seconds=None, callback=None):78        return self.connection.receive_message(79            self, number_messages=num_messages,80            visibility_timeout=visibility_timeout,81            attributes=attributes,82            wait_time_seconds=wait_time_seconds,83            callback=callback,84        )85    def delete_message(self, message, callback=None):86        return self.connection.delete_message(self, message, callback)87    def delete_message_batch(self, messages, callback=None):88        return self.connection.delete_message_batch(89            self, messages, callback=callback,90        )91    def change_message_visibility_batch(self, messages, callback=None):92        return self.connection.change_message_visibility_batch(93            self, messages, callback=callback,94        )95    def delete(self, callback=None):96        return self.connection.delete_queue(self, callback=callback)97    def count(self, page_size=10, vtimeout=10, callback=None,98              _attr='ApproximateNumberOfMessages'):99        return self.get_attributes(100            _attr, callback=transform(101                self._coerce_field_value, callback, _attr, int,102            ),...

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.