Skip to content

src/dazzle_back/events/dev_memory.py

Generated: 2026-03-30T04:38:46.041144+00:00

Commit: 615f1354a103

Language: python

Scope: dazzle_back

Purpose

In-Memory Event Bus Implementation for Testing.

DevBusMemory provides a simple in-memory implementation of the EventBus interface for use in unit tests. Events are stored in Python dicts with no persistence.

Features: - Fast, synchronous operations (no I/O) - Full EventBus interface compliance - U...

Key Exports

  • StoredEvent
  • ConsumerState
  • DevBusMemory
  • publish
  • subscribe
  • unsubscribe
  • ack
  • nack
  • replay
  • get_consumer_status
  • list_topics
  • list_consumer_groups
  • get_topic_info
  • process_pending
  • get_dlq_events
  • clear
  • get_all_events
  • event_count

Dependencies

Imports

  • __future__.annotations
  • asyncio
  • collections.defaultdict
  • collections.abc.AsyncIterator
  • dataclasses.dataclass
  • dataclasses.field
  • datetime.UTC
  • datetime.datetime
  • typing.Any
  • uuid.UUID
  • dazzle_back.events.bus.ConsumerNotFoundError
  • dazzle_back.events.bus.ConsumerStatus
  • dazzle_back.events.bus.EventBus
  • dazzle_back.events.bus.EventHandler
  • dazzle_back.events.bus.EventNotFoundError
  • dazzle_back.events.bus.NackReason
  • dazzle_back.events.bus.SubscriptionInfo
  • dazzle_back.events.envelope.EventEnvelope

Event Interactions

Emits: app.Order

Consumes: StoredEvent

Invariants and Assumptions

No invariants documented.

Tests

  • tests/unit/test_events.py
  • tests/unit/test_events_api.py

Notes

This page was automatically generated by tools/gen_reference_docs.py.