aeventkit

library2.1.0pypypi✓ verified 82d ago

aeventkit is an active Python library (current version 2.1.0) for building event-driven data pipelines. It leverages Pydantic v2 for robust data validation and AnyIO for asynchronous execution, making it suitable for modern, high-performance applications. The library focuses on clear separation of concerns with publishers, subscribers, and an event bus.

workflow data serialization http-networking

pip install aeventkit

github ↗package ↗

INSTALL

IMPORT

SIG · AEVENTKIT

aeventkit

workflowpythonv2.1.0

Install

3.5s avg

Import

—

Disk

87MB

Pass rate

8/ 10

Env Coverage8 / 10

glibc

3.9–3.13

musl

3.9–3.13

Install & Compatibility

Where this runs

tested against v2.1.0 · pip install

no network on importno background threads

Install × environment matrix

Each cell = how many times install + import succeeded across repeated harness runs. Partial = flaky.
glibc = Debian/Ubuntu slim · musl = Alpine Linux

musl

glibc

py 3.10

✓ —

✓ 3.65s

py 3.11

✓ —

✓ 3.45s

py 3.12

✓ —

✓ 3.35s

py 3.13

✓ —

✓ 3.45s

py 3.9

✕ build_error

87MB installed

● package 87MB

Code

Verified usage

Verified import paths — ran on the pinned version, not inferred.

Event

✓ from eventkit import Event

✗ from aeventkit import Event

This example demonstrates how to define a custom Event, create a Publisher and Subscriber, register them with an EventBus, and publish an event. It uses `anyio.run()` to execute the asynchronous main function, which is the standard entry point for aeventkit applications.

import anyio from aeventkit import Event, EventBus, EventSubscriber, EventPublisher from pydantic import Field class MyEvent(Event): message: str = Field(..., description="A simple test message") class MySubscriber(EventSubscriber): async def handle(self, event: MyEvent): print(f"Subscriber received: {event.message}") async def main(): bus = EventBus() subscriber = MySubscriber() publisher = EventPublisher() bus.register_subscriber(subscriber) bus.register_publisher(publisher) print("Publishing event...") await publisher.publish(MyEvent(message="Hello, aeventkit!")) await anyio.sleep(0.1) # Give subscriber a moment to process print("Event published and potentially handled.") if __name__ == "__main__": anyio.run(main)

Debug

Known issues

gotchaaeventkit is designed for asynchronous operations using AnyIO. All core components and interaction methods are `async`/`await` compatible. Attempting to call async methods from synchronous code without an event loop or proper execution context will result in errors.

fix

Ensure your application's entry point uses `anyio.run()` or integrates within an existing `asyncio` or `anyio` event loop. All event handlers and interactions should be `async` functions.

affects: >=2.0.0

breakingaeventkit strictly requires Pydantic v2 (>=2.0.0). Projects using Pydantic v1 will encounter `PydanticImportError` or validation issues due to API changes between Pydantic versions.

fix

Upgrade Pydantic to version 2 or higher (`pip install 'pydantic>=2.0.0' --upgrade`). Verify no other project dependencies are pinning Pydantic to v1.

affects: >=2.0.0

gotchaPublishers and Subscribers must be registered with an `EventBus` instance to communicate. Creating an `EventPublisher` or `EventSubscriber` alone does not make them active participants in the event stream.

fix

Always call `bus.register_publisher(publisher_instance)` and `bus.register_subscriber(subscriber_instance)` after initializing your bus and components.

affects: >=2.0.0

Upgrade

Version history

2.1.0latest on PyPI · released Jun 22, 2025

Audit

Dependencies

pydanticrequiredData validation and event model definition (requires >=2.0.0)

anyiorequiredAsynchronous concurrency backend (requires >=4.0.0)

logururequiredLogging (requires >=0.7.0)

richrequiredRich terminal output (requires >=13.0.0)

Agent activity

77 hits · last 30 days

node

OpenAI (training)

Resources

githubhttps://github.com/ib-api-reloaded/eventkit ↗

packagehttps://pypi.org/project/aeventkit/ ↗