Reference library

Streaming & messaging

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

3 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
…

16 0 Open

Streaming & messaging easy

Build a Streaming Messaging Helper in Python

Create a simple message stream class that stores recent messages, sends user messages, and retrieves history or latest messages with timestamps.

streaming deque dataclass

Python

from collections import deque
from dataclasses import dataclass
from datetime import datetime
import time


@dataclass
class Message:
    user: str
    text: str
    timestamp: str = ""

    def __post_init__(self):
        if not self.timestamp:
            self.timestamp = datetime.now().strftime("%H:%M:%S")


class…

13 0 Open

Streaming & messaging easy

How to Build a Message Stream Queue in Python

A beginner-friendly MessageStream class built on deque that sends messages one at a time, tracks unread counts, and records sent items.

queue deque streaming

Python

from collections import deque
import time


class MessageStream:
    def __init__(self, messages):
        self._queue = deque(messages)
        self._sent = []

    def send_next(self):
        if not self._queue:
            return None
        message = self._queue.popleft()
        self._sent.append(message)
     …

13 0 Open

Browse by section

Each section groups closely related Python snippets.