Reference library

Streaming & messaging

Kafka-style pub/sub, event consumers, async pipelines, and message-driven workflows.

3 matches

Dedupe processed message IDs in Python

Filters an inbox of messages by removing items whose IDs have already been processed, using a set for fast lookups.

from pathlib import Path
import json


def dedupe_processed_ids(inbox_file: Path, processed_file: Path) -> list:
    processed = set(json.loads(processed_file.read_text()))
    inbox = json.loads(inbox_file.read_text())
    deduped = [item for item in inbox if item["id"] not in processed]
    return deduped


if __nam…

13 0 Open

Streaming & messaging easy

Exactly Once Idempotent Consumer Store in Python

A mock key-value store that guarantees exactly-once processing by rejecting duplicate message keys in a message or event stream.

idempotency streaming deduplication

Python

from collections import defaultdict

class ExactlyOnceStore:
    def __init__(self):
        self.processed = defaultdict(set)
        self.data = {}

    def consume(self, key, value):
        if key in self.data:
            return False
        self.data[key] = value
        return True

    def get_processed_count…

15 0 Open

Streaming & messaging easy

How to Mock RabbitMQ Ack Nack Requeue in Python

A mock RabbitMQ channel and consumer that simulates ack, nack, and requeue handling for testing message processing logic without a broker.

rabbitmq testing mock

Python

import json
from collections import deque


class MockChannel:
    def __init__(self):
        self.acked = []
        self.nacked = []
        self.requeued = []

    def basic_ack(self, delivery_tag):
        self.acked.append(delivery_tag)

    def basic_nack(self, delivery_tag, requeue=False):
        self.nacked.…

15 0 Open

Browse by section

Each section groups closely related Python snippets.