Reference library

Streaming & messaging

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

3 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 Mock a Kafka Producer Batch Send in Python

Simulate a Kafka producer in Python that sends batched JSON events with mock partitions and latency for testing streaming pipelines without a real broker.

kafka mock streaming

Python

import json
import random
import time
from datetime import datetime


class MockKafkaProducer:
    def __init__(self, topic):
        self.topic = topic
        self.sent_messages = []

    def send(self, value, key=None):
        message = {
            "topic": self.topic,
            "key": key,
            "value"…

13 0 Open

Streaming & messaging medium

How to Read Redis Streams with XREADGROUP in Python

Read new messages from a Redis stream using a consumer group with XREADGROUP, handling JSON payloads and group creation.

redis streams consumer groups

Python

import redis
import json

def read_group_messages(stream_key, group_name, consumer_name, count=10):
    r = redis.Redis(host="localhost", port=6379, decode_responses=True)
    try:
        r.xgroup_create(stream_key, group_name, id="0", mkstream=True)
    except redis.exceptions.ResponseError:
        pass

    messag…

12 0 Open

Browse by section

Each section groups closely related Python snippets.