Reference library

Streaming & messaging

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

4 matches

Dead Letter Queue Failed Messages List Mock in Python

Implements a simple in-memory dead letter queue to collect, list, and retry failed messages, with JSON serialization for inspection in streaming pipelines.

dead-letter-queue messaging retry

Python

import json
from collections import deque


class Message:
    def __init__(self, message_id, payload, attempts=0):
        self.message_id = message_id
        self.payload = payload
        self.attempts = attempts

    def __repr__(self):
        return f"Message(id={self.message_id}, attempts={self.attempts})"


c…

16 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 easy

How to Serialize and Deserialize JSON Event Payloads in Python

Define an EventPayload class with custom to_json and from_json methods to convert event objects to JSON strings and back, using datetime parsing.

json serialization datetime

Python

import json
from datetime import datetime


class EventPayload:
    def __init__(self, event_id, event_type, timestamp, data):
        self.event_id = event_id
        self.event_type = event_type
        self.timestamp = timestamp
        self.data = data

    def to_json(self):
        return json.dumps({
          …

12 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)
  …

12 0 Open

Browse by section

Each section groups closely related Python snippets.