Reference library

Streaming & messaging

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

4 matches

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 Partition and Order Kafka-Style Messages by Key in Python

Group messages with the same key into ordered buckets using hashing and a defaultdict, mimicking Kafka partition ordering.

streaming partitioning kafka-pattern

Python

from dataclasses import dataclass
from collections import defaultdict

@dataclass
class Message:
    key: str
    content: str

def partition_and_order(messages, num_partitions=3):
    partitions = defaultdict(list)
    for msg in messages:
        partition_id = hash(msg.key) % num_partitions
        partitions[parti…

14 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

Streaming & messaging medium

Implement the Transactional Outbox Pattern with SQLite in Python

A Python implementation of the transactional outbox pattern using SQLite, ensuring atomic writes of order data and outbox events in a single transaction while supporting reliable message publishing and consumption.

outbox-pattern sqlite transactions

Python

import sqlite3
from dataclasses import dataclass
from datetime import datetime, timezone
import json

@dataclass
class Order:
    order_id: str
    amount: float
    status: str

class TransactionalOutbox:
    def __init__(self, db_path=":memory:"):
        self.conn = sqlite3.connect(db_path)
        self._create_tab…

17 0 Open

Browse by section

Each section groups closely related Python snippets.