Reference library

Streaming & messaging

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

2 matches

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…

19 0 Open

Browse by section

Each section groups closely related Python snippets.