English | Türkçe

EventFlow

Repository · Latest release · MIT License

EventFlow lets components in one Python process communicate through ordinary Python objects. It provides a synchronous event bus, an asyncio-native event bus, optional in-memory history, named channels, and explicit subscription lifecycle management.

Optional pipelines transform values through named stages and around middleware. Async streams compose lazy transformations with bounded buffering. Scopes, event envelopes and opt-in measurements let these pieces work together without requiring a framework or changing ordinary bus publication.

The library is for in-process delivery. It does not persist events, serialize them, or transport them between processes. Those responsibilities remain with the application or its messaging infrastructure.

EventFlow was extracted from the event subsystem of Vehicle Autonomy Core and is maintained here as a standalone library. The source project is not needed to install or use this package.

Contents

• Getting started
  • Install
  • Version
  • Publish your first event
  • Choose a bus
• Core concepts
  • Events and subscriptions
  • Filtering and delivery limits
  • Failures and results
• Working with events
  • History and replay
  • Waiting for an event
  • Periodic publishing
• Async applications
• Organizing channels
• Callback policies
• Pipelines and middleware
• Async pipelines and streams
• Subscription scopes
• Event context and measurements
• Application coordination
  • Event bridges and queues
  • Task and resource ownership
  • Concurrency and duplicate work
  • Retries, cooldowns and deadlines
  • Scheduling and bounded work
  • State, cache, services and policy
  • Operation metrics
• Security and operational boundaries
• Composing the components
• Lifecycle and reference
  • Defaults
  • Bus methods
  • Engine methods
  • Public types and errors
• Development and license

Getting started

Install

Python 3.10 or newer is required. From a checkout, install the source into the active environment:

python -m pip install .

After the PyPI release is published, install EventFlow with:

python -m pip install flowevent

flowevent is the PyPI distribution name. The library is called EventFlow, and Python code continues to import the eventflow package:

from eventflow import EventBus

Python 3.10 installs backports.strenum and exceptiongroup to supply features that are built into Python 3.11 and newer. Package metadata selects these dependencies only on Python 3.10.

The distribution includes inline type annotations and a py.typed marker, so type checkers can discover EventFlow's types when the package is installed.

Because the distribution installs files in the shared eventflow import package, installing another distribution that provides that package in the same environment can overwrite or conflict with these files. Use a separate virtual environment when both distributions are needed.

For editable development, install the optional tools and run the repository checks from the project root:

python -m pip install -e ".[dev]"
python -m unittest discover -s tests -v
ruff check .
pyright

The dev extra provides Ruff and Pyright. The tests use Python's standard unittest framework.

Version

The installed package exposes its version through eventflow.__version__:

from eventflow import __version__

print(__version__)  # 1.3.0

Publish your first event

An event is a value that describes something that has happened. It can be any Python object; a frozen dataclass is a useful choice when the event should be explicit and immutable.

from dataclasses import dataclass

from eventflow import EventBus


@dataclass(frozen=True)
class OrderPlaced:
    order_id: str


def send_confirmation(event: OrderPlaced) -> None:
    print(f"Order placed: {event.order_id}")


with EventBus[OrderPlaced]() as bus:
    with bus.subscribe(send_confirmation):
        result = bus.publish(OrderPlaced("order-42"))
        print(f"delivered={result.delivered}, failed={result.failed}")

Output:

Order placed: order-42
delivered=1, failed=0

The publisher creates an OrderPlaced value and passes it to the bus. The bus calls the subscribed handler before publish() returns. The inner context manager cancels the subscription when it ends; the outer one closes the bus. This makes the lifetime of both resources visible at the point of use.

The generic annotation in EventBus[OrderPlaced] helps type checkers and IDEs. It does not validate event types at runtime. Use a filter when a runtime type check is needed.

Choose a bus

	EventBus	AsyncEventBus
Handler	Synchronous callable	Async callable
Publish	Blocks until selected handlers finish	Await publish()
Execution context	Calling thread by default; optional executor	One owning asyncio event loop
Delivery order	Concurrent publisher calls may overlap	Sequential by default; optional concurrent delivery

Choose EventBus for synchronous code. By default, each handler runs on the thread that called publish(). An optional concurrent.futures.ThreadPoolExecutor can run handlers on worker threads, but the publishing call still waits for them. When buses share a pool, a callback that publishes through another bus using that same pool delivers the nested event inline on the active worker. This prevents worker starvation during cross-bus reentrant publication.

Publishing from multiple threads is safe. Two simultaneous publish calls may run their handlers at the same time, so handlers that update shared state must provide their own synchronization.

Choose AsyncEventBus when the application already uses asyncio and handlers are coroutines. Bus operations belong to the first running event loop that uses the bus. Do not share an instance across different loops. To publish from a worker thread, use publish_threadsafe(); see Async applications.

Core concepts

A bus stores active subscriptions and delivers each published event to the subscriptions whose filters match it. Subscriptions are independent: a handler can be cancelled without affecting other handlers or the bus.

Events and subscriptions

Events do not inherit from a library base class. Define the values that suit the application, then publish them with publish(event). Keep event objects small and give their fields names that describe the information a handler needs.

subscribe(callback) returns a Subscription for EventBus and an AsyncSubscription for AsyncEventBus. Keep the returned handle while the handler is needed. Call cancel() when it is no longer needed, or manage the handle with a context manager as in the first example. Cancellation is safe to repeat. The handle exposes id and active.

Filtering and delivery limits

A subscription can receive every event, events of one or more classes, or events that pass a predicate. A type check and predicate can be combined:

from dataclasses import dataclass

from eventflow import EventBus, EventFilter


@dataclass(frozen=True)
class PaymentReceived:
    account_id: str
    amount_cents: int


large_payment = EventFilter(
    event_type=PaymentReceived,
    predicate=lambda event: event.amount_cents >= 10_000,
)


def write_audit_record(event: PaymentReceived) -> None:
    print(f"Audit {event.account_id}: {event.amount_cents}")


with EventBus[PaymentReceived]() as bus:
    with bus.subscribe(write_audit_record, event_filter=large_payment):
        bus.publish(PaymentReceived("acct-7", 2_500))
        bus.publish(PaymentReceived("acct-7", 12_500))

Only the second payment reaches the handler. A type filter may also be a tuple of classes. Predicates are synchronous functions that receive the event and return a boolean, including when attached to an async bus. Async predicates and predicates that return awaitables are rejected because filtering itself is synchronous.

Use either an EventFilter or the direct event_type/predicate arguments on a bus method; do not combine the two forms. For a handler that should stop after its first match, use once=True or once(callback, ...). For a fixed number of matching deliveries, use times=N. These limits count matching events, not every event published to the bus.

Failures and results

The default ErrorPolicy.ISOLATE records handler failures and continues with the other selected handlers. Each call to publish() returns a PublishResult:

Field	Meaning
matched	Subscriptions selected for immediate delivery by this publish call
delivered	Handlers that completed successfully
failed / errors	Collected delivery or action failures
successful	True when no failures were collected

Events buffered for a subscription while replay is in progress are excluded from matched, even when its filter accepts them.

A synchronous on_error hook can report a failure without coupling the handler to a particular logging system:

from eventflow import EventBus, EventErrorContext


def report_failure(context: EventErrorContext[str]) -> None:
    print(f"{context.event}: {type(context.error).__name__}")


def persist_order(event: str) -> None:
    raise RuntimeError("storage is unavailable")


with EventBus[str](on_error=report_failure) as bus:
    bus.subscribe(persist_order)
    result = bus.publish("order-42")
    print(result.matched, result.delivered, result.failed)

Set error_policy=ErrorPolicy.RAISE when the caller should handle failures directly. The bus attempts delivery to the selected handlers, then raises an ExceptionGroup if failures were collected. That publish call does not return a PublishResult. The bus.stats property provides cumulative published, delivered, and failed counts.

These counters are not a per-publish invariant. published advances after history recording succeeds (or is skipped when history is disabled) and before callbacks run; delivered and failed are finalized afterward. An in-flight or cancelled async publish can leave them different. failed counts collected errors from handlers, filters, and actions, not only failed handlers.

To catch grouped failures with the same import on Python 3.10 and newer, import ExceptionGroup from eventflow.compatibility. On Python 3.11 and newer, that name refers to the built-in exception group class.

Working with events

History and replay

History is disabled by default. Set history to a positive capacity to keep recent events in a bounded, thread-safe MemoryEventHistory. When the buffer is full, each new event replaces the oldest one. History lives in memory; it is not a durable event log.

from eventflow import EventBus


with EventBus[str](history=3) as bus:
    for status in ("created", "paid", "packed", "shipped"):
        bus.publish(status)

    print(bus.query(limit=2))
    print(bus.latest())

Output:

('packed', 'shipped')
shipped

query() returns retained events in publication order. It accepts the same filter options as a subscription and an optional positive limit. With a limit, it returns the newest matching events in chronological order. latest() returns the newest match or None. Without history, query() returns an empty tuple and latest() returns None. History order reflects when events were recorded, not when callbacks start or finish. Async publish hooks can yield after recording, and separate publish calls may overlap.

An unfiltered query with limit reads only the requested recent window. A filtered query evaluates the predicate outside the history lock on a stable snapshot. It searches newest to oldest when limited, but a rare match near the oldest entry can still require scanning the full snapshot. On AsyncEventBus, this filtered lookup runs synchronously on the owning event loop. A rare match in a large history can therefore delay unrelated tasks; bound the retained history and keep predicates inexpensive when replay latency matters.

A new subscriber can receive recent events before it starts handling live ones. Pass replay=N to deliver up to the N newest retained events that match the subscription filter:

with EventBus[str](history=10) as bus:
    bus.publish("started")
    bus.publish("ready")

    with bus.subscribe(print, replay=2):
        pass

The handler prints started and ready as the subscription is created. Live events that arrive during replay are buffered per subscriber. replay_buffer_limit (default 1000) bounds this temporary queue. If the queue overflows, that subscription is cancelled and the overflow is reported using the bus's error policy.

To provide a custom history store, implement EventHistory: append(), latest(), query(), clear(), and __len__(). Implementations must synchronize all five operations and return latest()/query() results from a consistent snapshot. A history shared by multiple buses must synchronize across those buses too. The bus publication lock serializes append() only for that bus; history queries run outside it. Keep append() fast and non-blocking, and do not publish recursively through the same bus from it.

Bus clear() removes active subscriptions and wakes pending waiters; it does not erase history or stop periodic schedules. Call the returned schedule's cancel() to stop it, or close the bus to stop all schedules. To delete retained events, call clear() on the configured history object.

Waiting for an event

Use wait_for() when the caller needs one matching event as a value instead of registering a long-lived handler. It accepts the same filter options as subscribe() and an optional timeout. It returns the event, or None if the wait times out or the bus closes before a match. A synchronous wait blocks the calling thread. Timeouts must be finite, non-negative numbers; zero requests an immediate wait.

This async example schedules a publisher and then waits for its event. The wait registers its temporary subscription before the publisher task gets to run:

import asyncio

from eventflow import AsyncEventBus


async def main() -> None:
    async with AsyncEventBus[str]() as bus:
        publish_task = asyncio.create_task(bus.publish("ready"))
        event = await bus.wait_for(event_type=str, timeout=1.0)
        await publish_task
        print(event)


asyncio.run(main())

The output is ready. In synchronous code, use bus.wait_for(event_type=..., timeout=...); omit timeout to wait without a deadline. A configured on_timeout hook receives an EventTimeoutContext with the filter and timeout that expired.

Periodic publishing

publish_every(event, interval, times=None, immediately=True) publishes the same event repeatedly and returns a cancellable subscription. interval must be positive. The first publication happens immediately unless immediately=False. Set times to a positive integer for a finite schedule, or cancel the returned subscription to stop early. A bus allows at most max_schedules active schedules (default 64).

import time

from eventflow import EventBus


with EventBus[str]() as bus:
    bus.subscribe(print)
    schedule = bus.publish_every("heartbeat", interval=1.0, times=3)

    while schedule.active:
        time.sleep(0.05)

The handler receives three heartbeats. AsyncEventBus provides the awaited version and runs it on the owning loop:

import asyncio

from eventflow import AsyncEventBus


async def main() -> None:
    async with AsyncEventBus[str]() as bus:

        async def report(message: str) -> None:
            print(message)

        await bus.subscribe(report)
        schedule = await bus.publish_every("heartbeat", 0.05, times=3)
        while schedule.active:
            await asyncio.sleep(0.01)


asyncio.run(main())

Closing a bus cancels its schedules. With ErrorPolicy.RAISE, an error during a periodic publish stops that schedule. A schedule does not return each PublishResult; configure the bus on_error hook to observe handler, filter, or action failures. With ErrorPolicy.ISOLATE, the schedule continues and the publish result is otherwise discarded.

Async applications

AsyncEventBus accepts async handlers and async bus hooks. Its default DeliveryMode.SEQUENTIAL awaits matching handlers in subscription order. DeliveryMode.CONCURRENT awaits them together. Concurrent mode can be useful when handlers are independent, but their completion order is not guaranteed. Sequential ordering applies within one publish() call; separate publish calls may overlap. In sequential mode, a handler-raised asyncio.CancelledError propagates from publish() and stops delivery to later handlers. In concurrent mode, an independently cancelled handler task is recorded as a failed delivery in PublishResult. Cancelling the task that called publish() propagates in either mode and cancels in-flight handler work. Cancellation cannot undo side effects a handler produced before it was cancelled.

import asyncio

from eventflow import AsyncEventBus, DeliveryMode


async def main() -> None:
    async with AsyncEventBus[int](
        delivery_mode=DeliveryMode.CONCURRENT,
        max_concurrency=8,
    ) as bus:

        async def save_metric(value: int) -> None:
            print(f"Saved metric: {value}")

        async def notify_dashboard(value: int) -> None:
            print(f"Dashboard notified: {value}")

        await bus.subscribe(save_metric)
        await bus.subscribe(notify_dashboard)
        result = await bus.publish(42)
        print(f"delivered={result.delivered}")


asyncio.run(main())

Bus operations such as subscribe(), publish(), wait_for(), publish_every(), clear(), and close() are awaited. The publish_threadsafe(event) entry point is for a different thread: it returns a concurrent.futures.Future and schedules delivery on the owning loop.

Concurrent delivery is unbounded by default to preserve the existing behavior. Set max_concurrency=N with DeliveryMode.CONCURRENT to cap callback fan-out for each publish. When a callback finishes, its slot is reused immediately for the next subscriber instead of waiting for the rest of a fixed group. Overlapping publish calls each have their own limit; this does not bound how many publish calls the application creates. Choose a finite value when one event's fan-out could otherwise create too many tasks or use too many external resources.

import asyncio

from eventflow import AsyncEventBus


async def main() -> None:
    async with AsyncEventBus[str]() as bus:

        async def report(message: str) -> None:
            print(message)

        await bus.subscribe(report)  # binds the bus to this loop
        future = await asyncio.to_thread(
            bus.publish_threadsafe,
            "from a worker",
        )
        result = await asyncio.wrap_future(future)
        print(result.delivered)


asyncio.run(main())

Do not call future.result() on the event-loop thread: waiting there would block the loop needed to deliver the event.

Organizing channels

EventEngine creates synchronous buses lazily by channel name. Use it when several parts of an application need separate event channels with shared defaults:

from eventflow import EventEngine


def update_order_view(event: str) -> None:
    print(f"Order view updated: {event}")


with EventEngine(history=50) as engine:
    engine.subscribe("orders", update_order_view)
    result = engine.publish("orders", "order-42")
    print(result.delivered)

Channel names must be strings and are stripped and lowercased, so " Orders " and "orders" refer to the same channel. engine.channel(name) returns a channel, creating it when necessary. Use engine.add(name, bus) to register a bus with custom settings. remove(name) closes one channel; stop() closes every channel. An engine can own a bus instance under only one channel name at a time. A successful add() claims that bus for the engine; another engine cannot claim the same instance and receives ValueError. Removing or stopping the owner closes the bus. The engine exposes running and a sorted channel_names tuple. During stop(), both engines attempt every channel. If some closes fail, the failures are raised together and the engine retains those channels so a later stop() can retry cleanup.

AsyncEventEngine offers the same channel model for async buses. Its channel() and add() methods return directly. Publish, subscribe, once, wait, remove, start, and stop operations are awaited:

import asyncio

from eventflow import AsyncEventEngine


async def main() -> None:
    async with AsyncEventEngine(history=50) as engine:

        async def update_order_view(event: str) -> None:
            print(f"Order view updated: {event}")

        await engine.subscribe("orders", update_order_view)
        result = await engine.publish("orders", "order-42")
        print(result.delivered)


asyncio.run(main())

Engine defaults are passed to the buses the engine creates. EventEngine carries history, error policy, executor, and hooks; its optional executor must be a ThreadPoolExecutor. AsyncEventEngine carries history, error policy, delivery mode, max_concurrency, and async hooks.

Callback policies

CallbackSubscription is a lower-level utility for applying limits and hooks to one callback when it is invoked directly. It is independent of a bus and is not created by EventBus.subscribe(). A caller supplies its identifier, a cancellation function, and the callback:

from eventflow import CallbackSubscription


def handle_message(message: str) -> str:
    return message.upper()


policy = CallbackSubscription(
    identifier=1,
    cancel=lambda: None,
    callback=handle_message,
    max_calls=3,
    frequency_hz=2,
    on_success=lambda context: print(context.result),
)

print(policy.invoke("ready"))

invoke() applies the policy to a synchronous callback. For an asynchronous callback, set asynchronous=True and await invoke_async():

import asyncio

from eventflow import CallbackSubscription


async def handle_message(message: str) -> str:
    return message.upper()


async def main() -> None:
    policy = CallbackSubscription(
        identifier=2,
        cancel=lambda: None,
        callback=handle_message,
        asynchronous=True,
        timeout=1.0,
    )
    print(await policy.invoke_async("ready"))


asyncio.run(main())

The policy can limit accepted calls with once or max_calls, cap the rate with frequency_hz, apply a timeout, filter with a synchronous predicate, and be toggled with enable()/disable(). It supports before, success, error, timeout, and after hooks; each hook receives a CallbackContext with the event, result, error, and elapsed time. Disabled, filtered, and rate-limited events do not consume the call limit. An invocation that passes those checks consumes a call even if the callback later raises.

Use on_before(), on_success(), on_error(), on_timeout(), or on_after() to register hooks after construction. Calling the wrapper itself invokes the underlying callback directly and bypasses its policies; use invoke() or invoke_async() when those policies should apply.

A synchronous timeout is observational: Python cannot safely terminate a running thread, so the callback completes before an overrun is reported. An asynchronous timeout is cooperative: it requests cancellation and waits for the callback task to finish before raising CallbackTimeoutError. A callback that suppresses cancellation, or a hook that does not return, can extend the invocation beyond its configured budget. Python cannot force-stop a coroutine. A callback's own TimeoutError is distinct from CallbackTimeoutError.

Pipelines and middleware

Use Pipeline[T] for a sequence of transformations from T to T. Every stage receives the previous stage's result. The pipeline returns the final value and does not publish it automatically. An empty pipeline returns its input.

from dataclasses import dataclass, replace
from eventflow import Pipeline


@dataclass(frozen=True)
class Request:
    amount: int


pipeline = Pipeline[Request]()


@pipeline.stage("validate")
def validate(request: Request) -> Request:
    if request.amount < 0:
        raise ValueError("amount must be non-negative")
    return request


def double(request: Request) -> Request:
    return replace(request, amount=request.amount * 2)


pipeline.add("double", double, after="validate")


@pipeline.middleware("audit")
def audit(request, next_):
    print("before", request.amount)
    result = next_(request)
    print("after", result.amount)
    return result


print(pipeline.run(Request(4)))  # Request(amount=8)

stage() and middleware() register and return the original function. Its identity, name and direct call behavior remain intact. Decorator registration and add() use the same registry; neither wraps the registered function.

Operation	Behavior
add(name, callback, *, before=None, after=None)	Add a stage with optional ordering constraints.
stage(name, *, before=None, after=None)	Decorator equivalent of add.
remove(name)	Remove the stage and references to it; return its callback.
replace(name, callback)	Keep the name, ordering constraints and enabled state.
enable(name) / disable(name)	Change inclusion in future runs.
stage_names	Ordered names, including disabled stages.
add_middleware(...) / middleware(...)	Register around middleware in its own namespace.
remove_middleware, replace_middleware, enable_middleware, disable_middleware	Apply the corresponding registry operation to middleware.
middleware_names	Names from outermost to innermost, including disabled middleware.

Names are stripped, case-sensitive, non-empty and unique within each registry. before and after must reference existing names. Unknown references raise KeyError; duplicate names, empty names and ordering cycles raise ValueError. Invalid mutations leave the existing plan intact. Registration order resolves ties. Disabled stages retain their ordering constraints. Removing a stage removes references to it from the remaining stages.

Ordering is prepared when configuration changes. Each run takes one immutable execution snapshot; replacement or disabling during a run affects subsequent runs. Configuration is thread-safe and callbacks execute outside its lock. The snapshot does not copy payloads or make a callback's own mutable state safe for concurrent use.

Middleware wraps the entire pipeline: the first registered middleware is the outermost. It may call next_(value) once, transform its return value, catch a PipelineExecutionError to recover, or return directly to short-circuit. A continuation is valid only inside its middleware call and on the same thread or async task. A second call, deferred call or detached-task call raises RuntimeError. Existing bus actions and callback hooks retain their own semantics; they are not executed as pipeline middleware. Because middleware uses nested continuations, each pipeline accepts at most 64 registered middleware entries, including disabled entries. The limit is checked when adding middleware and avoids reaching Python's recursion limit during a run.

Stage failures stop later stages and raise PipelineExecutionError. Inspect name, phase, context and __cause__ to locate the original failure. The context is the value passed into the failing component. A synchronous stage or middleware returning an awaitable is rejected, and returned coroutine objects are closed. Python cannot forcibly stop a synchronous callback.

Async pipelines and streams

AsyncPipeline[T] has the same registry and decorators. Its callbacks return awaitables; await pipeline.run(value) executes on the first owning loop. Synchronous callbacks are never moved to threads. Concurrent runs use independent snapshots and continuations. CancelledError propagates into the active stage and is not converted into a delivery failure or retried. Stage and middleware results must be resolved values: returning another awaitable after the outer await (for example, forgetting await next_) is rejected and any returned coroutine is closed.

AsyncStream[T] owns an AsyncIterable[T]. Its transformations accept ordinary or awaitable callbacks, making both pipeline run methods directly usable as map functions. Mapping, filtering and batching are lazy and start no workers.

import asyncio
from eventflow import AsyncPipeline, AsyncStream


async def main() -> None:
    pipeline = AsyncPipeline[int]()

    @pipeline.stage("double")
    async def double(value: int) -> int:
        return value * 2

    async def source():
        for value in range(5):
            yield value

    stream = (
        AsyncStream(source())
        .map(pipeline.run)
        .filter(lambda value: value % 4 == 0)
        .buffer(2)
        .batch(2)
    )
    async with stream:
        async for batch in stream:
            print(batch)  # (0, 4), then (8,)


asyncio.run(main())

Operation	Result and ownership
AsyncStream(source)	A single-use stream on one owning loop.
map(callback)	Transform values sequentially; await awaitable results.
filter(predicate)	Keep accepted values; synchronous or awaitable predicate.
batch(size, *, partial=True)	Yield tuples; optionally include the final partial tuple.
buffer(capacity)	Add one producer and a bounded queue with backpressure.
aclose() / async with stream	Close the entire upstream chain and await owned cleanup.
closed	Whether new reads have stopped.

Batch sizes and buffer capacities must be positive integers; booleans, zero, fractions and infinity are rejected. A full buffer blocks its producer. There is room for capacity queued values plus one source value currently waiting to enter the queue. Values are never dropped. Source failures reach the consumer after earlier queued values; normal completion remains distinct from failure and cancellation. Capacity counts objects, not bytes; a small number of large payloads can still use substantial memory. Apply payload-size limits or an application-owned byte budget when events have highly variable sizes.

A stream is consumed once. A second iterator acquisition and concurrent reads are rejected. Use async with on the terminal stream when a consumer may break early or raise. Python's async for protocol does not notify a source when arbitrary consumer code uses break; aclose() supplies that explicit ownership boundary. It also closes an unstarted upstream chain. Closing from another task cancels an outstanding reader, then waits for source cleanup. Cancellation of the caller is propagated after owned cleanup finishes, even if cancellation is repeated. Sources and user cleanup still need to cooperate with cancellation; a non-returning source can delay closure. A callback running in the active read task cannot await that same terminal stream's aclose(); it raises RuntimeError to prevent a cleanup deadlock. The read's normal error path then closes the stream and its upstream chain.

Subscription scopes

Scopes group existing cancellation handles without changing what subscribe returns. add(handle) returns the same handle and ignores repeated registration of the same object. Separate scopes may own unrelated buses or callback policies.

from eventflow import EventBus, SubscriptionScope

with EventBus[int]() as bus, SubscriptionScope() as scope:
    handle = scope.add(bus.subscribe(print))
    bus.publish(42)
assert not handle.active

SubscriptionScope.close() cancels synchronous handles in reverse registration order. AsyncSubscriptionScope.aclose() awaits async handles in the same order and supports async with. Both accept duck-typed handles with the corresponding cancel() method. They attempt every handle, collect failures in the existing Python 3.10-compatible ExceptionGroup, and retain failed handles for retry. An async handle's cancel() must return an awaitable. The scope checks this when closing because add() cannot validate the result without cancelling the handle. Handle implementations should make repeated cancellation safe. A handle's own async cancellation is propagated after the other handles have been attempted. Other handle-owned BaseException interrupts also retain their type after all handles are attempted; additional failures are chained as an exception group. Closed scopes reject additions. Async cleanup is shared by concurrent close callers and finishes before caller cancellation is re-raised.

Scopes cancel registrations; they do not acquire ownership of the underlying bus or forcibly stop callbacks already running. Keep bus/engine context managers when those resources also need to close. The async scope itself belongs to one loop and rejects reentrant cleanup that would await itself.

Event context and measurements

publish(event) continues to accept raw objects. EventEnvelope[T] is an optional frozen container with event, generated id, correlation_id, source, UTC timestamp and metadata. A timezone-aware timestamp supplied in another zone is normalized to UTC. Metadata is copied into a read-only mapping, including after copy.deepcopy(); payloads and nested metadata values retain their own mutability. dataclasses.asdict() and pickle work when the event and metadata values themselves support those operations. Only unpickle data from trusted sources: Python pickle can execute code during loading. Use dataclasses.replace to transform an envelope's payload.

from dataclasses import replace
from eventflow import EventEnvelope, Pipeline, PipelineStats, correlation_context

stats = PipelineStats()
pipeline = Pipeline[EventEnvelope[int]](observer=stats)
pipeline.add("increment", lambda envelope: replace(envelope, event=envelope.event + 1))
with correlation_context("request-42"):
    event = EventEnvelope(1, source="example", metadata={"attempt": 1})
    result = pipeline.run(event)
print(result.event, result.correlation_id)  # 2 request-42
print(stats.snapshot.runs.successful)  # 1

correlation_context(id) is a normal context manager, usable in async functions too. current_correlation_id() reads the current identifier. An envelope uses an explicit correlation ID, then the bound context, then its own ID. Context is isolated across asyncio tasks and reset on exit. New threads and the bus executor do not automatically inherit it; use contextvars.copy_context in application code when propagation is required.

Pass observer=PipelineStats() or a synchronous PipelineObserver callable to either pipeline. Each completed run, stage and middleware produces a PipelineObservation(kind, name, elapsed, status, correlation_id, error). Statuses are success, error and cancelled; elapsed times are seconds. Bind correlation around the caller's entire run for consistent run and stage notifications. Middleware-local context applies only while that middleware is active. With no observer, clocks and observation objects are omitted.

stats.snapshot returns immutable runs, stages and middleware aggregates. Each ExecutionMetrics includes calls, successful, failed, cancelled, total_seconds and max_seconds. Counts describe actual execution, so a short-circuit does not count skipped stages. Run and middleware durations include nested stages; do not add them together. reset() clears aggregates. No payloads or traceback objects are retained by the collector. Metric keys are stage and middleware names; stable names keep the collector small. Dynamically generated names remain in the snapshot until reset() is called, so avoid unbounded user or request values as component names.

Observers are synchronous and execute inline, including in AsyncPipeline; they must be fast and non-blocking to avoid delaying the event loop. For logging or telemetry I/O, enqueue a compact observation into an application-owned bounded queue. An observer failure aborts an otherwise successful run with phase="observer". During an existing pipeline failure, observer errors are available as observer_errors on the original PipelineExecutionError, preserving its cause. Existing EventBusStats is unchanged.

Application coordination

EventFlow also provides small execution and coordination primitives for application code. Each is opt-in and has an explicit owner and shutdown method. They use ordinary Python callables, awaitables, futures, context managers and iterators, so they can be composed with buses, pipelines and streams without changing event delivery or stage ordering.

Component	Module	Purpose
AsyncEventQueue, LoopPortal	eventflow.bridges	Bridge synchronous bus delivery and external threads to an owned async loop.
AsyncTaskScope	eventflow.tasks	Own background asyncio tasks.
ResourceScope, AsyncResourceScope	eventflow.scopes	Run reverse-order, retryable resource cleanup.
KeyedConcurrencyLimiter, KeyedRateLimiter, SingleFlight	eventflow.concurrency	Bound keyed work, apply a token bucket, or share an in-flight operation.
LatestTaskCoordinator	eventflow.tasks	Keep only the newest result for each key.
AsyncStream.batch_within()	eventflow.stream	Emit a batch when its size or time window is reached.
BoundedWorkPool	eventflow.executors	Bound synchronous work submitted to a thread pool.
KeyedCoalescer, AsyncScheduler	eventflow.scheduling	Merge keyed updates and own scheduled async jobs.
RetryPolicy, CooldownRegistry, CircuitBreaker	eventflow.resilience	Coordinate retries and temporary failure handling.
Deadline	eventflow.timing	Share a monotonic time budget across nested operations.
StateMachine, VersionedValue	eventflow.state	Guard local state transitions and versioned values.
VersionedCache	eventflow.cache	Bound an in-memory LRU cache with versions and optional TTL.
ServiceRegistry	eventflow.registry	Register owner-managed services with cleanup.
PolicyEvaluator	eventflow.policy	Evaluate ordered synchronous rules.
OperationStats	eventflow.observability	Collect bounded-cardinality aggregates without retaining payloads.

These utilities are in-process building blocks. They do not persist state or coordinate independent processes. Import a component from eventflow for the lazy top-level API or from its module when a narrower import is preferred.

Event bridges and queues

AsyncEventQueue subscribes to a synchronous EventBus and exposes received events through receive() or async iteration. It has a fixed item capacity, accepts publishes from other threads, and supports drop_oldest, drop_newest and error overflow policies. The configured capacity includes events waiting to be delivered onto the owning loop. It allows one active consumer task at a time; another task may take over after the current consumer has finished. The queue remains bound to its original event loop. Closing the queue removes its subscription, allows already queued items to drain, then makes the next read raise EventQueueClosedError. If that loop has already closed, aclose() can still remove the synchronous bus subscription from a different running loop. stats() reports queue size, deliveries, drops and overflow errors. An error overflow raises QueueOverflowError inside the synchronous subscriber; the bus applies its configured ErrorPolicy to that callback failure.

import asyncio
from eventflow import AsyncEventQueue, EventBus

bus = EventBus[int]()


async def receive_one():
    async with AsyncEventQueue(bus, event_type=int, capacity=64, overflow="drop_oldest") as events:
        await asyncio.to_thread(bus.publish, 42)
        return await events.receive()


assert asyncio.run(receive_one()) == 42
bus.close()

LoopPortal owns a dedicated event-loop thread. Submit an awaitable to get a concurrent.futures.Future, or use call() from another thread when the caller needs a synchronous result. It copies the current contextvars context at submission time. close() rejects new work, cancels and waits for portal tasks, shuts down its default executor, and joins its thread. A blocking call() from the portal's own loop is rejected to prevent deadlock. The portal does not take ownership of an application's existing event loop. A call(timeout=...) timeout requests cancellation; it cannot force a cancellation-resistant operation to stop. An operation's own TimeoutError is propagated unchanged; the portal timeout is raised only when call()'s wait deadline expires. That operation remains owned by the portal until it actually finishes, and close() can time out while waiting for it. Work already running through asyncio.to_thread() cannot be stopped by task cancellation. A timed-out close() retains the portal loop thread while that worker finishes; call close() again to complete shutdown.

from eventflow import LoopPortal


async def load_value():
    return 42


with LoopPortal() as portal:
    future = portal.submit(load_value)
    assert future.result(timeout=2) == 42

Task and resource ownership

AsyncTaskScope owns tasks created with start(). Each task uses shutdown="cancel" or shutdown="drain"; closing the scope applies that task's policy, awaits cleanup, and rejects later starts. TaskHandle.join() waits for that task and returns its result. snapshot() reports active, completed, failed and cancelled work. A synchronous on_error callback observes failures not retrieved by a join. If aclose(timeout=...) times out, unfinished tasks remain owned and a later close can retry; a timeout never makes a still-running task disappear.

from eventflow import AsyncTaskScope


async def refresh_cache():
    pass


def report_task_error(_handle, _error):
    pass


async with AsyncTaskScope(on_error=report_task_error) as tasks:
    refresh = tasks.start(refresh_cache, name="cache-refresh", shutdown="cancel")
    await refresh.join()

ResourceScope and AsyncResourceScope own arbitrary cleanup operations. add(resource, close=..., owned=True) registers a method name or cleanup callable; defer(cleanup) registers a no-argument callback. Cleanup runs in reverse registration order, attempts every entry, and retains failures for a later retry. Use owned=False for a borrowed resource, or detach(resource) to transfer ownership out of the scope. pending_cleanup() reports entries still requiring cleanup. The async version awaits async closers and supports async with; cleanup is shared across concurrent close callers. A synchronous cleanup that reenters its own ResourceScope.close() gets a no-op from the nested call; the outer close retains responsibility for the pending count and cleanup results.

from eventflow import ResourceScope


class Resource:
    def close(self):
        pass


def flush_metrics():
    pass


with ResourceScope() as resources:
    resources.add(Resource())
    resources.defer(flush_metrics)

Concurrency and duplicate work

KeyedConcurrencyLimiter(max_concurrency) keeps a separate active-work limit for every key. Use async with limiter.acquire(key) for cancellation-safe release, or acquire_nowait(key) when the caller can handle immediate rejection. Idle keys are removed. aclose() rejects new acquisitions and wakes waiters. Queued acquisitions run in FIFO order, and acquire_nowait(key) returns False while a waiter is queued so it cannot bypass earlier work. This is a concurrency cap; it does not limit requests per second. If acquire_nowait(key) succeeds, release the slot with await limiter.release(key). stats() reports aggregate counts; key_stats(key) reports active permits and waiters for one key.

KeyedRateLimiter(rate, period, burst=...) is a per-key token bucket. await acquire(key) waits cancellably for a token; try_acquire(key) returns immediately. max_keys bounds retained keys and prune() removes idle state; if the key table is full and no idle bucket can be pruned, a new key is rejected. It has loop ownership and should be closed with aclose().

SingleFlight.run(key, factory) shares one in-flight awaitable among callers with the same key. Different keys run independently, one waiter's cancellation does not cancel shared work, and current waiters receive the same result or failure. The result is not cached after completion. Shared work continues if all waiters leave; it remains owned and visible through stats().active until completion. Call await flight.aclose() when the instance is no longer needed; it cancels and drains active work, then rejects new calls. Factories must cooperate with cancellation for shutdown to finish. A factory cannot close its own SingleFlight instance; that call raises RuntimeError. stats() must be called from the instance's owning event loop. LatestTaskCoordinator.submit(key, factory, on_result) cancels prior work by default and applies a result only if that submission is still the newest for the key. Generation checks also prevent a cancellation-resistant old task from publishing a stale result. on_result runs on the owning loop and may be synchronous or awaitable. Close the coordinator with aclose().

from eventflow import KeyedConcurrencyLimiter, SingleFlight

limiter = KeyedConcurrencyLimiter[str](max_concurrency=2)
flight = SingleFlight[str]()


async def fetch_account(account_id):
    return {"id": account_id}


async def load_for(account_id):
    async with limiter.acquire(account_id):
        return await flight.run(account_id, lambda: fetch_account(account_id))


await limiter.aclose()

Retries, cooldowns and deadlines

RetryPolicy offers synchronous run() and asynchronous run_async() with fixed or exponential backoff, a delay ceiling, optional jitter and an optional synchronous observation callback. Retries are explicit: without should_retry, a failure is raised after the first attempt even when max_attempts is larger. The synchronous predicate receives ordinary Exception instances and must return bool; cancellation is propagated and is never retried. A Deadline is a retry budget: it prevents attempts from starting after expiry and caps backoff delays. It does not interrupt an operation already in progress, and a successful operation may return after the deadline. Wrap an operation in a separate cancellation-aware timeout when its execution itself needs a hard time limit.

Deadline.after(seconds) uses a monotonic clock and exposes remaining and expired. child(seconds) creates a shorter budget that cannot outlive its parent. Passing None represents no deadline. The object communicates a time budget; it cannot forcibly stop synchronous code or a coroutine that suppresses cancellation.

CooldownRegistry stores bounded per-key monotonic expiry times. block() sets or extends a cooldown, remaining() reports its duration, and wait() / wait_async() can wait until it expires or is cleared. CircuitBreaker wraps synchronous call() or asynchronous call_async() operations and moves through closed, open and half_open states. It limits concurrent half-open probes and ignores cancellation as a counted failure. These primitives coordinate one process; they do not share state across workers or machines.

from eventflow import Deadline, RetryPolicy


class TemporaryError(Exception):
    pass


async def fetch_remote():
    return "response"


deadline = Deadline.after(5)
retry = RetryPolicy(max_attempts=3, initial_delay=0.05, jitter=0.1)
result = await retry.run_async(
    fetch_remote,
    should_retry=lambda error: isinstance(error, TemporaryError),
    deadline=deadline,
)

Scheduling and bounded work

AsyncScheduler owns one-shot and periodic asyncio jobs. Periodic jobs choose overlap="skip", "queue" or bounded "parallel" execution; the parallel limit is set with max_parallel. Queue mode retains at most one missed tick while a callback runs. Retain the returned ScheduledHandle when individual cancellation or joining is needed. aclose() stops new schedules, cancels active jobs and waits for their cleanup.

BoundedWorkPool runs synchronous callables in a thread pool with an explicit accepted-work bound: max_workers controls concurrent threads and max_pending controls queued work beyond those workers. submit() returns a standard future and rejects immediately when capacity is full. Close the pool with close() or its context manager; cancellation cannot stop a function that is already running.

KeyedCoalescer merges repeated values by a caller-supplied key and merge function, then delivers them from one owned worker at the configured interval. The first value for a key starts that key's interval. max_pending_keys bounds retained keys. Its worker starts on the first accepted push(), so an unused instance creates no background thread. The key function runs on the submitting thread before locking. The merge function also runs there, under the coalescer's condition lock, to preserve atomic updates for a key. Keep both functions fast and non-reentrant. Delivery runs on the worker. flush() delivers pending values immediately; close(flush=True) flushes before stopping, while flush=False discards pending values. Callback failures are reported through on_error and stats().

from eventflow import BoundedWorkPool


def parse_payload(payload):
    return payload


raw_payload = b"example"
with BoundedWorkPool(max_workers=4, max_pending=8) as workers:
    result = workers.submit(parse_payload, raw_payload).result(timeout=2)

AsyncStream.batch_within(max_items, max_delay, partial=True) emits when the first limit is reached. Its timer begins with the first item in a batch; it does not create one task per item. The final partial batch is emitted when partial=True and discarded when false. It retains the same single-consumer, event-loop ownership and explicit cleanup rules as other stream operations.

from eventflow import AsyncStream


async def source():
    for item in range(3):
        yield item


async def store_batch(batch):
    print(batch)


async with AsyncStream(source()).batch_within(100, 0.05) as batches:
    async for batch in batches:
        await store_batch(batch)

State, cache, services and policy

StateMachine(initial, transitions, terminal=..., guard=...) validates synchronous state transitions. transition() changes state and increments its version atomically; compare_and_transition(expected_version, target) returns None when another caller already changed the version. Observers run after the state update and outside the lock. VersionedValue offers the same snapshot/version and compare-and-set pattern for an arbitrary value. Its guard runs outside the lock and can be called again after a transition race, so it should be synchronous and free of side effects. A VersionedValue snapshot holds the value reference and does not copy mutable objects.

VersionedCache is a bounded in-memory LRU with optional TTL and byte budget. Provide getsizeof(key, value) when max_bytes is set. put() requires a version and rejects an older write while a newer version for that key remains resident. Expiry, eviction and removal also discard the version history, so an older write can be accepted after the entry leaves the cache. Use VersionedValue or durable state when versions must remain monotonic beyond cache residency. The cache is process-local and does not provide persistence or cross-process invalidation.

ServiceRegistry maps trimmed, non-empty names to explicitly registered services. Its cleanup callbacks must be synchronous. Every registration has an owner; close_owner(owner) closes that owner's resources in reverse registration order and retains failed cleanup for retry. An optional synchronous observer receives immutable registration and removal changes after mutation and outside the registry lock. Observer errors do not undo completed changes. The registry does not construct services or resolve dependency graphs.

PolicyEvaluator evaluates named synchronous rules in registration order. A rule returns "allow", "deny", "abstain", True, False or None; an explicit deny takes priority, then an allow applies, followed by the configured default (deny by default). Its PolicyDecision carries the decision and matching rule details. It is a small in-process rule combiner, not an identity provider or policy language.

from eventflow import StateMachine, VersionedValue

machine = StateMachine("queued", {"queued": {"running", "cancelled"}})
running = machine.transition("running")
value = VersionedValue("initial")
value.compare_and_set(value.snapshot().version, "ready")

Operation metrics

OperationStats is a thread-safe aggregate sink for calls, outcomes, elapsed time, queue depth, drops, retries, rejections, timeouts and shutdown time. Pass one instance to components that accept observer= to collect their measurements in the same immutable snapshot. max_operations bounds label cardinality; names beyond the limit share an other bucket. reset() clears the aggregates. Metrics retain counters only, not event payloads, callback results or exception objects. Keep operation names stable, and send telemetry outside the hot path through an application-owned bounded queue.

from eventflow import KeyedConcurrencyLimiter, OperationStats


async def update_account():
    pass


metrics = OperationStats(max_operations=64)
limiter = KeyedConcurrencyLimiter[str](1, observer=metrics)
async with limiter.acquire("account-42"):
    await update_account()
print(metrics.snapshot().operations["keyed_concurrency.acquire"].successful)
await limiter.aclose()

Security and operational boundaries

EventFlow invokes callbacks, predicates, middleware and stream sources in the application process with that process's permissions. It does not sandbox untrusted plugins; isolate such code with an operating-system process or container boundary and explicit permissions and resource limits. Async cancellation and callback timeouts are cooperative and do not provide a hard deadline for code that suppresses cancellation or blocks the event loop.

PipelineExecutionError.context and observer errors may reference application objects, including sensitive data. Avoid logging these objects wholesale; redact fields in the application's structured logging layer.

Composing the components

Use the components independently or connect their normal callable/iterator interfaces. A stage can be CallbackSubscription.invoke, a stream map can be Pipeline.run or AsyncPipeline.run, and a scope can own the handle returned by either bus. Bus publication does not implicitly invoke a pipeline or add an envelope.

Three complete, network-free examples are executable from the repository root:

python -m examples.integrated
python -m examples.bus_stream
python -m examples.coordination

integrated.py combines sync and async pipeline stages, decorators, callback policies, correlation, metrics, a buffered stream and an async bus. bus_stream.py uses a bounded application queue as a bus subscription and stream source. It owns one publisher task and joins it on exit; no task is created per event. coordination.py connects a synchronous bus, AsyncEventQueue, AsyncTaskScope, a keyed concurrency limit, explicit retries, a deadline and shared operation metrics.

Lifecycle and reference

Defaults

Option	EventBus	AsyncEventBus
history	None	None
error_policy	ErrorPolicy.ISOLATE	ErrorPolicy.ISOLATE
executor	None	Not used
delivery_mode	Not used	DeliveryMode.SEQUENTIAL
max_concurrency	Not used	None (unbounded)
replay_buffer_limit	1000	1000
max_schedules	64	64
shutdown_timeout	5 seconds	Not used
actions or direct on_* hooks	None	None

history accepts a positive capacity or an EventHistory implementation. Pass either an EventBusActions/AsyncEventBusActions object or individual on_before, on_after, on_error, and on_timeout callbacks, not both. Hooks on EventBus are synchronous; hooks on AsyncEventBus are async.

Use close() when a bus is no longer needed; context managers do this automatically. A closed bus rejects new subscriptions and publications with EventBusClosedError. clear() removes subscriptions, leaves history intact, does not stop periodic schedules, and completes pending wait_for() calls with None on both bus types. A synchronous wait also returns None when the bus closes.

The synchronous bus waits up to shutdown_timeout for periodic workers to stop. If a worker does not stop in time, EventShutdownTimeoutError is raised. The worker is not forcibly terminated, so close() can be retried after the blocking handler finishes. Async close cancels and awaits periodic tasks. Closing a bus does not wait for active publish() calls: a callback that has already started may finish after close() returns.

Bus methods

Method	Behavior
subscribe(callback, *, event_filter=None, event_type=None, predicate=None, once=False, times=None, replay=0)	Register a handler. Supply either event_filter or direct filter arguments.
once(callback, *, event_filter=None, event_type=None, predicate=None, replay=0)	Register for the first matching event.
publish(event)	Deliver an event and return a PublishResult, unless the error policy raises.
wait_for(*, event_filter=None, event_type=None, predicate=None, timeout=None)	Return one matching event or None.
publish_every(event, interval, *, times=None, immediately=True)	Start a cancellable periodic schedule.
latest(...) / query(..., limit=None)	Read retained history. These queries are synchronous on both bus types.
clear() / close()	Clear subscriptions (schedules continue) or close the bus (schedules stop). Both are awaited on AsyncEventBus.

On AsyncEventBus, await subscribe(), once(), publish(), wait_for(), publish_every(), clear(), close(), and subscription cancellation. publish_threadsafe(event) is called from another thread without await and returns a concurrent.futures.Future.

Engine methods

EventEngine provides channel(), add(), remove(), publish(), subscribe(), once(), wait_for(), publish_every(), start(), and stop(). channel() and add() return directly.

Engine subscribe() accepts event_filter, event_type, predicate, once, times, and replay. Engine once() accepts the filter options and replay. The async engine exposes the same typed options for async handlers.

AsyncEventEngine provides the same operations for async buses. channel() and add() return directly; publishing, subscription, waiting, removal, start, and stop are awaited. Both engines support context managers; the async engine uses async with.

After a successful stop(), an engine can be reused: a later channel() or add() call creates or registers new open channels and marks it running again.

Public types and errors

The package exports the following public names:

Area	Public names
Buses	EventBus, AsyncEventBus, BaseEventBus
Engines	EventEngine, AsyncEventEngine
Subscriptions	Subscription, AsyncSubscription, CallbackSubscription
Filters and history	EventFilter, EventHistory, MemoryEventHistory
Results and modes	PublishResult, EventBusStats, ErrorPolicy, DeliveryMode
Actions and contexts	EventBusActions, AsyncEventBusActions, EventErrorContext, EventTimeoutContext, CallbackContext
Coordination and lifecycle	AsyncEventQueue, AsyncTaskScope, ResourceScope, AsyncResourceScope, LoopPortal, TaskHandle
Concurrency and scheduling	KeyedConcurrencyLimiter, KeyedConcurrencyKeyStats, KeyedRateLimiter, SingleFlight, LatestTaskCoordinator, AsyncScheduler, KeyedCoalescer, BoundedWorkPool
State and resilience	StateMachine, VersionedValue, VersionedCache, ServiceRegistry, RegistryChange, PolicyEvaluator, RetryPolicy, CooldownRegistry, CircuitBreaker, Deadline
Stream and metrics	AsyncStream, OperationStats, OperationMetrics, PipelineStats, EventEnvelope
Errors	EventBusError, EventBusClosedError, InvalidEventHandlerError, CallbackTimeoutError, EventShutdownTimeoutError

EventFilter accepts an event class or tuple of classes and an optional predicate. MemoryEventHistory(capacity=1000) is the built-in bounded, thread-safe history; capacity reports its configured size. Buses expose closed, subscriber_count, stats, history, and error_policy. Engines expose running and channel_names. Subscription handles expose id and active; CallbackSubscription also exposes enabled.

Error	Raised when
EventBusError	A bus operation or schedule capacity cannot be completed.
EventBusClosedError	A publish or registration is attempted after close.
InvalidEventHandlerError	A handler or hook does not match the bus's sync/async mode.
CallbackTimeoutError	A callback exceeds its configured time budget.
EventShutdownTimeoutError	A synchronous periodic worker outlives shutdown_timeout.
EventQueueClosedError	A read is attempted after a closed queue has drained.
QueueOverflowError	An AsyncEventQueue configured with overflow="error" is full.
LoopPortalClosedError	Work is submitted after a portal has closed.
CircuitOpenError	A circuit is open or its half-open probe limit has been reached.
StateTransitionError	A state transition is not allowed by the configured graph.

BaseEventBus is the abstract base for custom bus implementations. EventHistory is the interface for custom history stores. Most applications can work entirely with EventBus or AsyncEventBus.

Development and license

The project supports Python 3.10 and newer and uses setuptools as its build backend. The compatibility dependencies are installed only on Python 3.10. Development setup and checks are described under Install. Run python -m benchmarks.benchmark_replay from the repository root to compare history latest()/query() calls through the history and bus APIs, along with bounded sync and async replay setup. The standard-library benchmark defaults to 100,000 retained events and can be adjusted with --events, --replay, --iterations, and --samples.

Additional repeatable measurements use python -m benchmarks.benchmark_publish, python -m benchmarks.benchmark_pipeline, python -m benchmarks.benchmark_stream and python -m benchmarks.benchmark_runtime. The stream benchmark also compares burst-input batch_within(64, 60) against native size-only batching; this isolates batch coordination overhead and does not measure sparse-source timer latency. The runtime-load benchmark compares sync and async publisher contention, async max_concurrency values, callback error rates, periodic schedule resources, and traced memory over a configurable event soak. It defaults to five samples, 500 publishes per publisher in each sample, and 100,000 soak events; use --iterations, --publisher-counts, --schedule-counts, --samples or --soak-operations to adjust the workload. Use --output results.json to retain the environment, throughput, p50/p95/p99 latency and separately traced peak allocations. Stream rates count items per whole-stream operation. Runtime memory samples use tracemalloc, not process RSS, and include event history when configured. Local microbenchmarks are not deployment guarantees. Recorded results and methodology live under benchmarks/results.

Copyright © 2026 TahsinCr. EventFlow is licensed under the MIT License. See LICENSE and CHANGELOG.