Reference library

Streaming & messaging

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

5 matches

At Most Once Fire-and-Forget Mock in Python

A Python mock that enforces send() is called at most once and records the arguments for verification.

class FireForgetMock:
    def __init__(self):
        self._calls = 0
        self._last_args = None
        self._last_kwargs = None

    def send(self, *args, **kwargs):
        if self._calls > 0:
            raise RuntimeError("send() called more than once")
        self._calls += 1
        self._last_args = args
…

15 0 Open

Streaming & messaging easy

How to Build a Materialized View Updater Consumer Mock in Python

A mock consumer that queues change events and triggers refresh callbacks to simulate materialized view updates.

dataclasses deque mocking

Python

import time
from collections import deque
from dataclasses import dataclass, field
from typing import Callable, Deque, Optional


@dataclass
class MaterializedViewUpdater:
    """Mock updater that consumes change events and refreshes a view."""
    refresh: Optional[Callable[[str], None]] = None
    queue: Deque[tuple…

14 0 Open

Streaming & messaging easy

How to Implement Publish-Subscribe Fanout with Multiple Subscribers in Python

Create a simple publish-subscribe system in Python that broadcasts messages to multiple subscriber callbacks for a given topic.

pubsub messaging events

Python

import time

class PubSub:
    def __init__(self):
        self.subscribers = {}

    def subscribe(self, topic, callback):
        if topic not in self.subscribers:
            self.subscribers[topic] = []
        self.subscribers[topic].append(callback)

    def publish(self, topic, message):
        if topic in sel…

14 0 Open

Streaming & messaging easy

How to Implement a Tumbling Window Counter in Python

Count events that fall within a fixed-size sliding time window using a deque and pruning logic.

streaming window aggregation

Python

from collections import deque
import time


class TumblingWindowCounter:
    def __init__(self, window_size_seconds):
        self.window_size = window_size_seconds
        self.window = deque()

    def add_event(self, timestamp):
        self.window.append(timestamp)

    def count(self, current_time):
        while…

14 0 Open

Streaming & messaging easy

How to Implement an In-Memory Pub/Sub System in Python

This code implements a simple in-memory publish/subscribe system in Python, allowing topics, callbacks, and message broadcasting.

pubsub event-driven design-pattern

Python

class PubSub:
    def __init__(self):
        self.topics = {}

    def subscribe(self, topic, callback):
        if topic not in self.topics:
            self.topics[topic] = []
        self.topics[topic].append(callback)
        return lambda: self.unsubscribe(topic, callback)

    def unsubscribe(self, topic, callb…

18 0 Open

Browse by section

Each section groups closely related Python snippets.