Reference library

Streaming & messaging

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

5 matches

How to Aggregate Periodic Snapshot Data in Python

Generates mock snapshot data and groups values into periods to compute average aggregates with Python's standard library.

aggregation snapshots streaming

Python

import random
from collections import defaultdict

def snapshot_aggregate(n=10, period=3):
    data = defaultdict(list)
    for i in range(n):
        key = f"item_{i % period}"
        data[key].append(random.randint(1, 100))
    return dict(data)

def aggregate_periodic(snapshots, period=3):
    result = {}
    for …

14 0 Open

Streaming & messaging medium

How to Encode and Decode Avro Data in Python (Roundtrip)

Serialize a Python dict to Avro binary bytes and decode it back using the fastavro-compatible avro library.

avro serialization encode

Python

import io
import json
from avro.schema import parse
from avro.io import DatumWriter, DatumReader, BinaryEncoder, BinaryDecoder

def avro_roundtrip(schema_json, data):
    schema = parse(json.dumps(schema_json))
    bytes_writer = io.BytesIO()
    encoder = BinaryEncoder(bytes_writer)
    writer = DatumWriter(schema)
 …

14 0 Open

Streaming & messaging medium

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 CQRS projector read model update in Python

Build a CQRS projector class that maintains denormalized read models by applying domain events in a mock order-processing service.

cqrs projector read-model

Python

from dataclasses import dataclass, field
from typing import Dict, List, Optional


@dataclass
class OrderReadModel:
    order_id: str
    customer_name: str
    total: float
    status: str = "pending"
    items: List[Dict] = field(default_factory=list)

    def apply_event(self, event_type: str, payload: Dict) -> Non…

11 0 Open

Streaming & messaging medium

Implement the Transactional Outbox Pattern with SQLite in Python

A Python implementation of the transactional outbox pattern using SQLite, ensuring atomic writes of order data and outbox events in a single transaction while supporting reliable message publishing and consumption.

outbox-pattern sqlite transactions

Python

import sqlite3
from dataclasses import dataclass
from datetime import datetime, timezone
import json

@dataclass
class Order:
    order_id: str
    amount: float
    status: str

class TransactionalOutbox:
    def __init__(self, db_path=":memory:"):
        self.conn = sqlite3.connect(db_path)
        self._create_tab…

22 0 Open

Browse by section

Each section groups closely related Python snippets.