At Least Once with Idempotent Consumer in Python

import threading
import time
import uuid
from collections import Counter


class IdempotentConsumer:
    def __init__(self):
        self.processed = set()
        self._lock = threading.Lock()

    def consume(self, message_id, payload):
        with self._lock:
            if message_id in self.processed:
                return "duplicate"
            self.processed.add(message_id)
        # Simulate work
        time.sleep(0.01)
        return f"processed: {payload}"


class MockProducer:
    def __init__(self, consumer):
        self.consumer = consumer

    def send_at_least_once(self, payload, times=3):
        results = Counter()
        for _ in range(times):
            msg_id = str(uuid.uuid4())
            if self.consumer.consume(msg_id, payload) == "duplicate":
                results["duplicate"] += 1
            else:
                results["processed"] += 1
        return dict(results)


if __name__ == "__main__":
    consumer = IdempotentConsumer()
    producer = MockProducer(consumer)
    # Each unique message is processed exactly once, retries are duplicates
    first = producer.send_at_least_once("hello")
    second = producer.send_at_least_once("hello")
    print("First batch:", first)
    print("Second batch:", second)
    print("Total unique processed:", len(consumer.processed))