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 easy

Dedupe processed message IDs in Python

Filters an inbox of messages by removing items whose IDs have already been processed, using a set for fast lookups.

deduplication streaming json

Python

from pathlib import Path
import json


def dedupe_processed_ids(inbox_file: Path, processed_file: Path) -> list:
    processed = set(json.loads(processed_file.read_text()))
    inbox = json.loads(inbox_file.read_text())
    deduped = [item for item in inbox if item["id"] not in processed]
    return deduped


if __nam…

13 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 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

Browse by section

Each section groups closely related Python snippets.