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.
Python code
49 linesimport 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
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
- Use `queue.Queue` for thread-safe operations
- 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
More from Cloud + Python
- Build a URL Shortener Client with Python medium
- Create a Cloud Storage Helper Class in Python easy
- Create a Data Helper Class for Beginners in Python easy
- Cross Account Role Chaining Mock Credentials in Python medium
- Exponential Backoff with Jitter for Cloud API Calls in Python medium
- Generate Mock CloudFormation Stack Events in Python easy
Keep learning
Related tutorials and quizzes for this topic.