Reference library

Streaming & messaging

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

5 matches

Build a Streaming Messaging Helper in Python

Create a simple message stream class that stores recent messages, sends user messages, and retrieves history or latest messages with timestamps.

streaming deque dataclass

Python

from collections import deque
from dataclasses import dataclass
from datetime import datetime
import time


@dataclass
class Message:
    user: str
    text: str
    timestamp: str = ""

    def __post_init__(self):
        if not self.timestamp:
            self.timestamp = datetime.now().strftime("%H:%M:%S")


class…

13 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 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 Implement a Priority Queue for Messages in Python

Build a message priority queue with heapq and dataclasses that pops messages by priority, using sequence numbers to keep insertion order.

priority-queue heapq dataclass

Python

import heapq
from dataclasses import dataclass, field
from typing import Any

@dataclass(order=True)
class Message:
    priority: int
    sequence: int = field(compare=False)
    content: str = field(compare=False)

class PriorityQueue:
    def __init__(self):
        self._heap = []

    def push(self, priority: int,…

15 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.