How to Implement an Exactly-Once Deduplication Store in Python
Implement a Python class that deduplicates keys exactly once, tracking first-seen timestamps and duplicate counts.
Python code
35 linesfrom datetime import datetime
from typing import Any, Hashable
class ExactlyOnceStore:
def __init__(self) -> None:
self._seen: set[Hashable] = set()
self._first_seen: dict[Hashable, datetime] = {}
self._counts: dict[Hashable, int] = {}
def add(self, key: Hashable, value: Any = None) -> bool:
if key in self._seen:
self._counts[key] += 1
return False
self._seen.add(key)
self._first_seen[key] = datetime.now()
self._counts[key] = 1
return True
def is_unique(self, key: Hashable) -> bool:
return key in self._seen and self._counts[key] == 1
def summary(self) -> dict[str, Any]:
return {
"total_keys": len(self._seen),
"duplicates": {str(k): v for k, v in self._counts.items() if v > 1},
}
if __name__ == "__main__":
store = ExactlyOnceStore()
keys = ["apple", "banana", "apple", "cherry", "banana", "apple"]
results = [store.add(k) for k in keys]
print(results)
print(store.summary())
Output
[True, True, False, True, False, False]
{'total_keys': 3, 'duplicates': {'apple': 3, 'banana': 2}}
How it works
The ExactlyOnceStore uses a set to track keys that have been seen before. When add is called, it checks the set; if the key is new, it records the first-seen timestamp and sets the count to 1. If the key already exists, it increments the count and returns False, indicating a duplicate. The is_unique method returns True only for keys seen exactly once. The summary method provides a dictionary of total unique keys and a breakdown of keys that have duplicates, which is useful for auditing or monitoring in a distributed system.
Common mistakes
- Assuming keys are strings; keys can be any hashable type, so conversion to string in summary might hide type info.
- Not handling thread safety; this implementation is not thread-safe and will fail with concurrent access.
- Using `datetime.now()` may not be monotonic; prefer `time.monotonic()` for ordering.
- Forgetting to clear state when store grows large; memory usage grows with number of keys.
Variations
- Use a TTL (time-to-live) to expire keys after a certain time.
- Implement using `collections.Counter` for simpler duplicate counting.
Real-world use cases
- In a microservices architecture, deduplicating incoming webhook events where each event has a unique ID.
- In a data pipeline, ensuring that a batch of records is processed only once per unique key (e.g., user_id) for idempotent writes.
- In a message queue consumer, preventing duplicate processing of messages that might be delivered more than once.
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.