Reference library

Streaming & messaging

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

4 matches

How to Build a Materialized View Updater Consumer Mock in Python

A mock consumer that queues change events and triggers refresh callbacks to simulate materialized view updates.

import time
from collections import deque
from dataclasses import dataclass, field
from typing import Callable, Deque, Optional


@dataclass
class MaterializedViewUpdater:
    """Mock updater that consumes change events and refreshes a view."""
    refresh: Optional[Callable[[str], None]] = None
    queue: Deque[tuple…

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

Browse by section

Each section groups closely related Python snippets.