Reference library

Streaming & messaging

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

4 matches

Batch Consume Process Commit Pattern in Python

A mock batch processor that accumulates items in a queue, processes full batches, commits successful or failed results, and flushes remaining items.

streaming batch-processing queues

Python

import random
import threading
import time
from collections import deque


class MockBatchProcessor:
    def __init__(self, process_func, commit_func, batch_size=5):
        self.queue = deque()
        self.batch_size = batch_size
        self.process_func = process_func
        self.commit_func = commit_func

    de…

14 0 Open

Streaming & messaging easy

How to Build a Mock Change Data Capture Event Stream in Python

Generate a deterministic list of mock CDC events with event IDs, stream positions, payloads, and timestamps for testing streaming pipelines.

cdc mock event-stream

Python

from itertools import count
from random import choice, randint, seed
from datetime import datetime, timedelta

seed(42)  # Make output deterministic
event_types = ["INSERT", "UPDATE", "DELETE"]
table_names = ["users", "orders", "products", "payments"]
counter = count(1)

def mock_cdc_event(stream_index: int) -> dict:
…

12 0 Open

Streaming & messaging easy

How to Wrap Message Attributes in a CloudEvent with Python

Create a minimal CloudEvent dataclass that wraps arbitrary message attributes into a JSON envelope, matching CloudEvents 1.0 spec.

cloudevents messaging dataclasses

Python

import json
from dataclasses import dataclass, field, asdict
from typing import Any, Dict
from datetime import datetime, timezone


@dataclass
class CloudEvent:
    message_attributes: Dict[str, Any] = field(default_factory=dict)

    def wrap(self, event_id: str, source: str, event_type: str, data: Any):
        self…

13 0 Open

Streaming & messaging hard

Mock Protobuf Binary Encoding in Python

Demonstrates a minimal protobuf-like binary encoding and decoding of an event dataclass using varints and length-delimited fields in pure Python.

protobuf binary-encoding varint

Python

import struct
from dataclasses import dataclass


@dataclass
class Event:
    id: int
    user_id: int
    action: str

    def encode(self) -> bytes:
        # Mock protobuf-like binary encoding using varint and length-delimited fields
        buf = bytearray()
        # field 1: varint id (tag = (1 << 3) | 0 = 8)
  …

11 0 Open

Browse by section

Each section groups closely related Python snippets.