Reference library

Streaming & messaging

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

5 matches

How to Implement an Outbox Table Poll Publisher in Python

This code simulates an outbox pattern with a class that polls for pending records and publishes them as JSON messages, removing only those that are due.

outbox polling messaging

Python

import time
import json
from dataclasses import dataclass, asdict
from datetime import datetime, timedelta

@dataclass
class OutboxRecord:
    id: int
    topic: str
    payload: dict
    created_at: datetime

class OutboxPollPublisher:
    def __init__(self, poll_interval_seconds=1):
        self.poll_interval = poll…

11 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 medium

How to Track Session Windows with Gap Timeout in Python

A Python class that groups events into sessions, closing a session when the gap between events exceeds a timeout threshold.

session-window streaming timeout

Python

import time

class SessionWindow:
    """Track sessions with a gap timeout (mock)."""
    
    def __init__(self, timeout_seconds=5):
        self.timeout = timeout_seconds
        self.session_start = None
        self.last_event_time = None
        self.event_count = 0
        self.events = []
    
    def add_event…

13 0 Open

Streaming & messaging medium

Implement a retry queue with visibility timeout in Python

This code simulates a message queue with a visibility timeout, allowing messages to be retried if not deleted before the timeout expires.

queue retry visibility-timeout

Python

import time
from collections import deque


class SimpleQueue:
    def __init__(self, visibility_timeout=2):
        self.queue = deque()
        self.in_flight = {}
        self.visibility_timeout = visibility_timeout

    def send(self, message):
        self.queue.append(message)

    def receive(self):
        if …

13 0 Open

Streaming & messaging medium

Mock Watermark Late Event Side Output in Python

Simulates watermarking in a streaming pipeline by classifying events as on-time or late using timestamps and delays.

watermark streaming side output

Python

from datetime import datetime, timedelta
from typing import List, Tuple


def watermark_mock(
    events: List[Tuple[datetime, str]], watermark_delay: timedelta, max_delay: timedelta
) -> Tuple[List[Tuple[datetime, str]], List[Tuple[datetime, str]]]:
    """Simulate watermarking: events arriving on time vs. late by ch…

11 0 Open

Browse by section

Each section groups closely related Python snippets.