Inbox pattern consumer dedupe mock in Python
Implements a mock inbox consumer that deduplicates incoming messages by ID, with automatic eviction of old seen IDs to prevent unbounded memory growth.
Python code
46 linesimport json
from collections import deque
from dataclasses import dataclass, field
from hashlib import sha256
from typing import Any
@dataclass
class InboxConsumer:
max_seen: int = 1000
seen_ids: set = field(default_factory=set)
seen_history: deque = field(default_factory=deque)
def _mark_seen(self, message_id: str) -> None:
if message_id not in self.seen_ids:
self.seen_history.append(message_id)
if len(self.seen_history) > self.max_seen:
evicted = self.seen_history.popleft()
self.seen_ids.discard(evicted)
def consume(self, message: dict[str, Any]) -> None:
message_id = message.get("id", "")
message_id = sha256(json.dumps(message).encode()).hexdigest() if not message_id else message_id
if message_id in self.seen_ids:
print(f"DEDUPE id={message_id} payload={message['payload']}")
return
self._mark_seen(message_id)
print(f"CONSUMED id={message_id} payload={message['payload']}")
if __name__ == "__main__":
inbox = InboxConsumer(max_seen=3)
messages = [
{"id": "a1", "payload": "first"},
{"id": "a2", "payload": "second"},
{"id": "a1", "payload": "first (duplicate)"},
{"payload": "no-id-message"},
{"payload": "no-id-message (duplicate)"},
{"id": "a3", "payload": "third"},
{"id": "a4", "payload": "fourth"},
{"id": "a1", "payload": "first (old, should pass now)"},
]
for msg in messages:
inbox.consume(msg)
Output
CONSUMED id=a1 payload=first
CONSUMED id=a2 payload=second
DEDUPE id=a1 payload=first (duplicate)
CONSUMED id=<sha256-hash> payload=no-id-message
DEDUPE id=<sha256-hash> payload=no-id-message (duplicate)
CONSUMED id=a3 payload=third
CONSUMED id=a4 payload=fourth
CONSUMED id=a1 payload=first (old, should pass now)
How it works
The seen_ids set provides O(1) membership tests, while seen_history as a deque tracks insertion order for eviction. _mark_seen appends only new IDs and removes the oldest when max_seen is exceeded, using discard to avoid errors. For messages without an ID, a SHA-256 hash of the JSON payload generates a deterministic ID, enabling deduplication of identical payloads. When a duplicate arrives, the consumer prints DEDUPE and skips processing, emulating an at-least-once inbox pattern.
Common mistakes
- Using a plain list instead of a set for O(n) lookups at scale
- Forgetting to evict old IDs, causing unbounded memory growth
- Hashing the message string instead of the JSON canonical form, leading to false dedupes
- Resetting seen IDs on restart when persistence is needed
Variations
- Use a time-based expiration (e.g., datetime timestamps) instead of a fixed count
- Persist seen IDs to Redis with TTL for cross-instance deduplication
Real-world use cases
- Deduplicating webhook events from payment providers that may retry deliveries.
- Preventing duplicate processing of IoT sensor readings in a streaming pipeline.
- Filtering duplicate user actions in an analytics ingestion service.
Sponsored
More from System design patterns
- Build a BFF (Backend for Frontend) Mock Aggregator in Python medium
- Builder pattern for mocking complex objects in Python easy
- Circuit Breaker Pattern in Python: Closed, Open, and Half-Open States medium
- Create a Data Helper Class in Python easy
- Domain Driven Design Aggregate Root Example in Python medium
- Facade Pattern in Python with Mock Simplification medium
Keep learning
Related tutorials and quizzes for this topic.