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.

Easy Python 3.9+ Aug 9, 2026 Microservices patterns 12 views 0 copies

Python code

44 lines
Python 3.9+
```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

stdout
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

  1. Use a TTL-based cache like Redis `SETNX` with expiry to deduplicate events across multiple service instances
  2. 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

Run this sample

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

Open editor

More from Microservices patterns

Related tutorials and quizzes for this topic.