python-eventmodel 0.0.0a3

Build type-safe, event-driven Python apps with zero boilerplate.

EventModel

Documentation is available at docs.eventmodel.app

Build type-safe, event-driven Python apps with zero boilerplate. Inspired by FastAPI and SQLModel, EventModel turns Pydantic classes into self-routing events. Just type-hint your inputs to subscribe and use return statements to emit. Pure, clean, and heavily typed.

Installation

uv add python-eventmodel

Quick Start

Define your events, type-hint your handlers, and let the framework handle the routing.

from eventmodel import App, EventModel, StopEvent

app = App()

# 1. Define events with embedded routing metadata
class UserCreated(EventModel, topic="user.events.created"):
    user_id: int
    email: str

class SendWelcomeEmail(EventModel, topic="email.queue.outbound"):
    target_email: str
    body: str

# 2. Write pure domain logic. 
# The input type dictates the subscription topic.
# The return type dictates the emission topic.
@app.service()
async def process_new_user(event: UserCreated) -> SendWelcomeEmail:
    print(f"Processing new user: {event.user_id}")
    return SendWelcomeEmail(
        target_email=event.email, 
        body="Welcome!"
    )

@app.service()
async def send_welcome_email(event: SendWelcomeEmail) -> StopEvent:
    print(f"Sending email to {event.target_email} with body: {event.body}")
    return StopEvent()


app.publish(UserCreated(user_id=123, email="[e-mail]"))
app.run()

Synchronous Handlers

EventModel is fully compatible with synchronous code. You can use standard def functions for your handlers instead of async def, and the framework will automatically execute them safely without blocking the main async event loop.

@app.service()
def sync_process_new_user(event: UserCreated) -> SendWelcomeEmail:
    print(f"Sync processing user: {event.user_id}")
    return SendWelcomeEmail(
        target_email=event.email, 
        body="Welcome from sync!"
    )

Core Concepts

1. Self-Routing Events: By subclassing EventModel and specifying a topic, your data model intrinsically knows where it belongs on the message broker.
2. Implicit Subscriptions: Handlers use the @service.service() decorator. The framework reads the type-hint of the input argument to determine which topic to subscribe to.
3. Implicit Emissions: Handlers return initialized event objects. The framework automatically intercepts these returns and publishes them to their respective topics.
4. Clean Architecture: Domain functions remain entirely pure. They take an EventModel as input and return an EventModel as output—no broker logic mixed in.
5. Modular Routing: The framework supports Service instances that can be merged into a master App via app.include().