Reference library

Streaming & messaging

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

8 matches
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

Dead Letter Queue Failed Messages List Mock in Python

Implements a simple in-memory dead letter queue to collect, list, and retry failed messages, with JSON serialization for inspection in streaming pipelines.

dead-letter-queue messaging retry
Python
import json
from collections import deque


class Message:
    def __init__(self, message_id, payload, attempts=0):
        self.message_id = message_id
        self.payload = payload
        self.attempts = attempts

    def __repr__(self):
        return f"Message(id={self.message_id}, attempts={self.attempts})"


c…
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 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
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 medium

How to Stream Join Windowed Mock Topics in Python

Simulates two message topics and joins their events when timestamps fall within a sliding time window using Python generators and deques.

streaming join generator
Python
import itertools
import random
import time
from collections import deque
from dataclasses import dataclass, field

@dataclass
class Event:
    key: str
    value: int
    timestamp: float = field(default_factory=time.time)

def generate_topic(prefix, keys, start_time):
    while True:
        yield Event(
            …
14 0 Open
Streaming & messaging easy

Implement a FIFO Message Queue in Python with deque

This code implements a FIFO (first-in-first-out) message queue class using Python's collections.deque, providing enqueue, dequeue, peek, and size operations.

queue deque fifo
Python
from collections import deque

class MessageQueue:
    def __init__(self):
        self.queue = deque()

    def enqueue(self, message):
        self.queue.append(message)
        print(f"Enqueued: {message}")

    def dequeue(self):
        if self.is_empty():
            print("Queue is empty, cannot dequeue.")
    …
14 0 Open
Streaming & messaging easy

Sliding Window Average with Deque in Python

Computes the running average of a sliding window over streaming numbers using a collections.deque for O(1) pop-left operations.

sliding-window deque streaming
Python
from collections import deque

class SlidingAverage:
    def __init__(self, window_size):
        self.window_size = window_size
        self.window = deque()
        self.total = 0

    def add(self, value):
        self.window.append(value)
        self.total += value
        if len(self.window) > self.window_size:
…
13 0 Open

Browse by section

Each section groups closely related Python snippets.

Streaming & messaging — Python code examples

What you will find here

This page collects streaming & messaging snippets — short, copy-ready Python you can paste into our free online IDE and run without installing anything. Each sample includes a plain-English explanation and the full source code.

Samples vs tutorials and challenges

Samples are quick reference — one concept per page. For step-by-step teaching, use our Python tutorials. To test yourself, try quizzes or coding challenges. Clean up style with the Python formatter.