sqspy package

Submodules

sqspy._base module

class sqspy._base.Base(**kwargs)[source]

Bases: object

Base class initialisation to setup aws credentials.

To make use of instance roles when deploying to AWS infrastructure, leave the aws_* keys blank (None).

Parameters

  • aws_access_key_id (str) – AWS access key credential.

  • aws_secret_access_key (str) – AWS access key credential.

  • profile_name (str) – Local AWS credential profile name.

  • region_name (str) – AWS region for resources.

  • endpoint_url (str) – Custom endpoint URL for usage.

QUEUE_VISIBILITY_TIMEOUT: str = '600'

Message’s visibility timeout in seconds. See Visibility Timeout in Amazon Simple Queue Service Developer Guide for more information.

create_queue(name: str, attributes: Dict[str, Any])[source]

Create a Queue resource.

For more information, check SQS.ServiceResource.create_queue()

Parameters

  • name (str) – Sent as the QueueName to the boto3 method.

  • attributes (dict[str,str]) – Same as parameter Attributes to create_queue()

Returns

A Queue resource

Return type

SQS.Queue

get_or_create_queue(queue_data: Dict[str, str], create_queue: bool = True)[source]

Fetch or create the sqs Queue resource from boto3.

Also tries to create the queue resource with the configured credentials as dictated by the create_queue parameter if the resource was not located.

Parameters

  • queue_data (dict[str,str]) –

    Dictionary referencing parameters for the queue to be retrieved or created.

    The keys for the data are: name, url and visibility_timeout. The visibility_timeout defaults to QUEUE_VISIBILITY_TIMEOUT.

  • create_queue (bool) – Force creation of queue resource on AWS. Default is True

Returns

An Queue resource on success, None otherwise.

Return type

SQS.Queue or None

get_queue(queue_data: Dict[str, str])[source]

Retrieve the Queue resource based on provided parameters.

Parameters

queue_data (dict[str,str]) – Same as used for get_or_create_queue()

Returns

sqspy.consumer module

class sqspy.consumer.Consumer(queue_name: Optional[str] = None, queue_url: Optional[str] = None, **kwargs)[source]

Bases: sqspy._base.Base

Message consumer/worker.

Parameters

  • queue_name (str) – Optional queue name.

  • queue_url (str) – Optional queue url, according to AWS guidelines.

  • queue (SQS.Queue) – Optional queue resource.

  • visibility_timeout (str) – Message visibility timeout in seconds, but as a string value. Defaults to QUEUE_VISIBILITY_TIMEOUT

  • error_queue (str) – Name for error queue, when messages were not consumed successfully.

  • error_queue_url (str) – Queue url as per AWS guidelines, for the error queue.

  • error_visibility_timeout (str) – Same as visibility_timeout but for error queue.

  • create_queue (bool) – Set to False if the queue should not be created in case it does not exist. The default is True.

  • create_error_queue (bool) – Same as create_queue but for error queue. The default is True.

  • poll_interval (int) – Polling interval between messages. Defaults to poll_interval.

  • message_attribute_names (list[str]) – List of attributes for message to fetch. See SQS.Message.message_attributes.

  • wait_time (int) – Time to wait (in seconds) when fetching messages. Defaults to wait_time.

  • force_delete (bool) – Whether to delete the message from queue before handling or not. Defaults to False.

  • max_messages_count (int) – Maximum message count when fetching from the queue. Defaults to max_messages_count.

  • attribute_names (list[str]) – Attributes to be retrieved along with message when fetching. See more at: SQS.Queue.receive_messages()

Raises

ValueError – At least one of queue, queue_url or queue_name has to be provided.

property error_queue

The Queue resource for when messages were not processed correctly.

Return type

SQS.Queue

abstract handle_message(body, attributes, messages_attributes)[source]

Method representing the handling of messages retrieved from queue.

Parameters

  • body (Any) – The body retrieved from the queue after passing through a json deserialiser.

  • attributes (dict) –

    A map of the attributes requested from queue when fetching messages.

    See more at: SQS.Message.attributes

  • messages_attributes (dict) –

    Strucutred metadata as retrieved from the queue.

    See more at: SQS.Message.message_attributes

Return type

None

Raises

NotImplementedError – If not overridden in a subclass.

listen()[source]

Method that triggers listening for messages, and forwards to handle_message().

This is a blocking call.

max_messages_count: int = 1

Upper limit of message count when fetching from queue.

poll_interval: int = 60

Time between continuous fetch from queue (in seconds).

poll_messages()[source]

Poll the queue for new messages.

The polling happens as per the poll_interval specified, and the message fetch timeout is set as per the value in wait_time.

Returns

A list of message resources.

Return type

list[SQS.Message]

property queue

The connected Queue resource.

Return type

SQS.Queue

property queue_name

Base name of the connected Queue resource.

Return type

str

wait_time: int = 0

Wait time when fetching message from queue (in seconds).

sqspy.producer module

class sqspy.producer.Producer(queue_name: Optional[str] = None, queue_url: Optional[str] = None, **kwargs)[source]

Bases: sqspy._base.Base

Message producer.

Parameters

  • queue_name (str) – Optional queue name.

  • queue_url (str) – Optional queue url, according to AWS guidelines.

  • queue (SQS.Queue) – Optional queue resource.

  • visibility_timeout (str) – Message visibility timeout in seconds, but as a string value. Defaults to QUEUE_VISIBILITY_TIMEOUT

  • create_queue (bool) – Set to False if the queue should not be created in case it does not exist. Default value is True.

Raises

ValueError – At least one of queue, queue_url or queue_name has to be provided.

publish(message: Any, **kwargs)[source]

Method to publish message to queue.

The message should be json serializable. The other arguments can be sent as named parameters. More information is available at SQS.Queue.send_message().

Parameters

message (Any) – The message body.

Returns

Dictionary of attributes as per AWS guidelines. Check: send_message().

Return type

dict

property queue

See queue

property queue_name

See queue_name

Module contents