Idempotent Consumer Event Processing in Python

Track processed event IDs to skip duplicates and count event types for a reliable, idempotent consumer.

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

idempotency events microservices set

Python code

30 lines

Python 3.9+

import json
from collections import defaultdict

class EventProcessor:
    def __init__(self):
        self.processed_ids = set()
        self.counts = defaultdict(int)

    def process_event(self, event):
        event_id = event["id"]
        if event_id in self.processed_ids:
            return {"status": "skipped", "reason": "already processed", "event_id": event_id}
        self.processed_ids.add(event_id)
        event_type = event["type"]
        self.counts[event_type] += 1
        return {"status": "processed", "event_id": event_id, "type": event_type}

if __name__ == "__main__":
    processor = EventProcessor()
    events = [
        {"id": "evt-1", "type": "order_created"},
        {"id": "evt-2", "type": "payment_received"},
        {"id": "evt-1", "type": "order_created"},
        {"id": "evt-3", "type": "order_created"},
        {"id": "evt-2", "type": "payment_received"},
    ]
    for evt in events:
        result = processor.process_event(evt)
        print(json.dumps(result))
    print("Final counts:", dict(processor.counts))

Output

stdout

{"status": "processed", "event_id": "evt-1", "type": "order_created"}
{"status": "processed", "event_id": "evt-2", "type": "payment_received"}
{"status": "skipped", "reason": "already processed", "event_id": "evt-1"}
{"status": "processed", "event_id": "evt-3", "type": "order_created"}
{"status": "skipped", "reason": "already processed", "event_id": "evt-2"}
Final counts: {'order_created': 2, 'payment_received': 1}

How it works

The EventProcessor keeps a set of seen event IDs to guarantee that each event is handled only once, even if duplicates arrive. Before processing, it checks membership in processed_ids; if present, it returns a skipped status and avoids double-counting or side effects. For new events, it records the ID, increments the type counter using defaultdict, and returns a processed status. This pattern mirrors how production consumers use a persistent store (e.g., Redis or a database) to make event handling idempotent across retries and replays.

Common mistakes

Using a list instead of a set for processed_ids causes O(n) lookups and slow performance with many events.
Forgetting to mark an event as processed before executing side effects can lead to duplicate work if an exception occurs.
Storing IDs in memory only loses idempotency if the process restarts — use a durable store for production.
Not including a timestamp or version in the processed record can make debugging duplicate issues harder.

Variations

Use Redis with SETNX or a database unique constraint to persist processed IDs across processes.
Use a `functools.lru_cache` or a `dict` keyed by event ID to store processing results for quick retrieval.

Real-world use cases

Consuming events from a message queue like Kafka where retries can deliver the same message multiple times.
Processing webhook callbacks from payment providers that might send duplicate notifications for the same transaction.
Handling event-driven state updates in a microservice where replay of a log could re-trigger the same action.

Idempotent Consumer Event Processing in Python

Python code

Output

How it works

Common mistakes

Variations

Real-world use cases

More from Microservices patterns

Tutorials

Quizzes

Python code

Output

How it works

Common mistakes

Variations

Real-world use cases

More from Microservices patterns

Keep learning

Tutorials

Quizzes