Reference library

Streaming & messaging

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

3 matches

How to Implement a Priority Queue for Messages in Python

Build a message priority queue with heapq and dataclasses that pops messages by priority, using sequence numbers to keep insertion order.

priority-queue heapq dataclass

Python

import heapq
from dataclasses import dataclass, field
from typing import Any

@dataclass(order=True)
class Message:
    priority: int
    sequence: int = field(compare=False)
    content: str = field(compare=False)

class PriorityQueue:
    def __init__(self):
        self._heap = []

    def push(self, priority: int,…

15 0 Open

Streaming & messaging easy

How to Simulate a Micro-Batch Interval Trigger in Python

A dataclass-based mock that emits batch numbers at fixed intervals, mimicking a micro-batch streaming scheduler for testing and development.

streaming mock dataclass

Python

import time
from dataclasses import dataclass, field
from typing import List, Callable


@dataclass
class MicroBatchTriggerMock:
    batch_interval_seconds: float = 0.5
    max_batches: int = 5
    _batches_emitted: int = 0
    _next_emit_time: float = field(init=False, default=0)

    def start(self, on_batch: Callab…

13 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.