How to Deduplicate Events in Python with SHA256 Hashing
Build an event deduplicator that identifies duplicate inbox messages using SHA256 hashes and tracks duplicate counts per event type.
Python code
44 lines```python
import hashlib
import json
from collections import defaultdict
class EventDeduplicator:
def __init__(self):
self.seen_hashes = set()
self.duplicate_counts = defaultdict(int)
def process_event(self, event):
event_key = f"{event['event_id']}:{event['timestamp']}"
event_hash = hashlib.sha256(event_key.encode()).hexdigest()
if event_hash in self.seen_hashes:
self.duplicate_counts[event["event_type"]] += 1
return False
self.seen_hashes.add(event_hash)
return True
def get_duplicate_stats(self):
return dict(self.duplicate_counts)
if __name__ == "__main__":
deduplicator = EventDeduplicator()
mock_events = [
{"event_id": "evt_001", "event_type": "message", "timestamp": 1719000000, "data": "Hello"},
{"event_id": "evt_002", "event_type": "message", "timestamp": 1719000001, "data": "World"},
{"event_id": "evt_001", "event_type": "message", "timestamp": 1719000000, "data": "Hello"},
{"event_id": "evt_003", "event_type": "notification", "timestamp": 1719000002, "data": "Alert"},
{"event_id": "evt_003", "event_type": "notification", "timestamp": 1719000002, "data": "Alert"},
]
processed = []
for event in mock_events:
if deduplicator.process_event(event):
processed.append(event)
print(f"Processed: {len(processed)} unique events")
print(f"Duplicates: {deduplicator.get_duplicate_stats()}")
print(json.dumps(processed, indent=2))
Output
Processed: 3 unique events
Duplicates: {'message': 1, 'notification': 1}
[
{
"event_id": "evt_001",
"event_type": "message",
"timestamp": 1719000000,
"data": "Hello"
},
{
"event_id": "evt_002",
"event_type": "message",
"timestamp": 1719000001,
"data": "World"
},
{
"event_id": "evt_003",
"event_type": "notification",
"timestamp": 1719000002,
"data": "Alert"
}
]
How it works
The EventDeduplicator class generates a SHA256 hash from the event's ID and timestamp, which uniquely identifies the event payload. On each call to process_event, the class checks if the hash already exists in the seen_hashes set; if it does, the event is classified as a duplicate and the counter for its event type is incremented. New unique events get added to the set, making subsequent lookups O(1) on average. The get_duplicate_stats method returns a plain dictionary so callers can easily inspect the distribution of duplicates across event types.
Common mistakes
- Hashing only the event_id without including the timestamp, which can cause false positives if IDs are reused
- Storing the full event object in memory instead of just the hash, defeating the memory efficiency of this approach
- Forgetting that hashes don't detect mutations—if the data changes but ID and timestamp stay the same, it's still marked as a duplicate
Variations
- Use a TTL-based cache like Redis `SETNX` with expiry to deduplicate events across multiple service instances
- Store the hash with the event ID in a database table with a unique constraint for durable deduplication across restarts
Real-world use cases
- Preventing double-processing of incoming webhooks or Kafka messages when consumers retry after network failures.
- Deduplicating event payloads in a message queue backlog so the same business event isn't handled multiple times.
- Tracking duplicate deliveries from a notification service to avoid sending the same alert to a user twice.
Sponsored
More from Microservices patterns
- BFF aggregation pattern: combine multiple service responses in Python easy
- Backward Compatible Schema Evolution in Python medium
- Bulkhead Thread Pool per Service Mock in Python medium
- CQRS with Separate Read and Write Repositories in Python medium
- Cache-Aside Pattern in Python: Per-Service Mock easy
- Consumer Driven Contract Pact Mock in Python medium
Keep learning
Related tutorials and quizzes for this topic.