How to Mock AWS SQS Send Receive Delete in Python

Build an in-memory mock of the SQS send, receive, and delete message flow for local testing.

Medium Python 3.9+ Aug 9, 2026 Cloud + Python 14 views 0 copies

Python code

49 lines
Python 3.9+
import json
from collections import deque
from uuid import uuid4


class MockSQSQueue:
    def __init__(self, name):
        self.name = name
        self._messages = deque()
        self._in_flight = {}

    def send_message(self, body, attributes=None):
        message_id = str(uuid4())
        message = {
            "MessageId": message_id,
            "Body": body,
            "Attributes": attributes or {},
        }
        self._messages.append(message)
        return {"MessageId": message_id}

    def receive_message(self, max_number=1, visibility_timeout=30):
        received = []
        for _ in range(min(max_number, len(self._messages))):
            message = self._messages.popleft()
            self._in_flight[message["MessageId"]] = message
            received.append(message)
        return received

    def delete_message(self, receipt_handle):
        # In this mock, receipt_handle is the MessageId
        if receipt_handle in self._in_flight:
            del self._in_flight[receipt_handle]
            return True
        return False


if __name__ == "__main__":
    queue = MockSQSQueue("orders")
    queue.send_message(json.dumps({"order_id": 1, "item": "laptop"}))
    queue.send_message(json.dumps({"order_id": 2, "item": "mouse"}))

    msgs = queue.receive_message(max_number=2)
    for msg in msgs:
        print(f"Received: {msg}")

    deleted = [queue.delete_message(m["MessageId"]) for m in msgs]
    print(f"Deleted: {deleted}")
    print(f"Remaining in queue: {len(queue._messages)}")

Output

stdout
Received: {'MessageId': '1234abcd-...', 'Body': '{"order_id": 1, "item": "laptop"}', 'Attributes': {}}
Received: {'MessageId': '5678efgh-...', 'Body': '{"order_id": 2, "item": "mouse"}', 'Attributes': {}}
Deleted: [True, True]
Remaining in queue: 2

How it works

This mock uses a deque to store messages and a dict to track in-flight messages, mimicking SQS's send/receive/delete flow. The send_message method appends a message with a UUID and optional attributes. receive_message pops messages from the queue and moves them to the in-flight dict, simulating visibility timeout. delete_message removes in-flight messages by their ID. This lets you test queue logic locally without AWS credentials or network calls.

Common mistakes

  • Forgetting to remove messages from the in-flight dict on successful processing
  • Not handling empty queue receive calls gracefully
  • Confusing MessageId with ReceiptHandle in production SQS
  • Using a list instead of deque, which makes popping from the front O(n)

Variations

  1. Use `queue.Queue` for thread-safe operations
  2. Add a visibility timeout re-queue mechanism for unprocessed messages

Real-world use cases

  • Unit testing a worker that polls, processes, and deletes SQS messages without AWS
  • Local development of event-driven services that consume SQS queues in a CI environment
  • Simulating queue behavior for load testing or integration tests in a sandbox

Sponsored

Run this sample

Open the browser IDE to tweak the example and see results without installing anything.

Open editor

More from Cloud + Python

Related tutorials and quizzes for this topic.