ducto.events module

Typed event emitter for credit lifecycle events.

Events are emitted by CreditManager after each store operation. The emitter is optional — inject into CreditManager constructor, no-op if omitted.

Usage:

from ducto.events import CreditEventEmitter

emitter = CreditEventEmitter()
emitter.on("credits.deducted", lambda event: print(event))
manager = CreditManager(store=store, emitter=emitter)

class ducto.events.CreditEvent(, type: str, timestamp: datetime, user_id: str, data: dict[str, Any] | None = None)

Bases: BaseModel

A typed credit lifecycle event.

data : dict[str, Any] | None

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

timestamp : datetime

type : CreditEventType

user_id : str

class ducto.events.CreditEventEmitter

Bases: object

Typed pub/sub event emitter for credit events.

Handlers are registered by event type string and called synchronously when the event is emitted. No-op when no handlers are registered.

clear_all() → None

Remove all handlers for all types.

clear_type(event_type: str) → None

Remove all handlers for a specific type.

emit(event: CreditEvent) → None

Emit an event to all registered handlers.

off(event_type: str, handler: Callable[[CreditEvent], None]) → None

Remove a previously registered handler.

class ducto.events.CreditEvent(, type: str, timestamp: datetime, user_id: str, data: dict[str, Any] | None = None)​

data : dict[str, Any] | None​

model_config = {}​

timestamp : datetime​

type : CreditEventType​

user_id : str​

class ducto.events.CreditEventEmitter​

clear_all() → None​

clear_type(event_type: str) → None​

emit(event: CreditEvent) → None​

off(event_type: str, handler: Callable[[CreditEvent], None]) → None​

on(event_type: str, handler: Callable[[CreditEvent], None]) → None​