Events

Nitro has two separate systems that are easy to conflate but serve distinct purposes:

Routing — HTTP request/response. Entity methods decorated with @post(), @get(), etc. are auto-registered as handlers. dispatch_action() calls them directly. No PubSub involved.
Events (PubSub) — Fire-and-forget, one-to-many messaging. subscribe(), publish(), publish_sync() are used for SSE streaming, notifications, and side effects.

Routing

Entity methods decorated with @post(), @get(), @put(), or @delete() are automatically registered in the handler registry. The framework calls them directly — no event bus in the middle.

from nitro import Entity, post, get, action

class Counter(Entity, table=True):
    count: int = 0

    @post()
    async def increment(self, request=None, amount: int = 1):
        self.count += amount
        self.save()
        return self.count

    @get()
    def status(self):
        return {"count": self.count, "id": self.id}

Decorators work on standalone functions too:

from nitro import post, get

@get(prefix="api")
def health():
    return {"status": "ok"}

@post(prefix="utils")
async def process(data: str = ""):
    return {"processed": data.upper()}

Handler Types

The registry handles all four Python callable forms transparently:

Type	Behavior
Sync function	Called directly, result returned
Async function	Awaited, result returned
Sync generator	Collected into a list
Async generator	Collected into a list

Calling Handlers

Use action() in templates to generate Datastar action strings that the browser sends back as HTTP requests:

counter = Counter.get("c1")
Button("+1", on_click=action(counter.increment, amount=1))
# →

Events (PubSub)

PubSub is for side effects and SSE streaming — not routing. When an entity completes an action, it can publish an event. Zero or more subscribers receive it independently.

from nitro.events import subscribe, publish, publish_sync

@subscribe("order.placed")
async def send_notification(msg):
    order = msg.data
    await EmailService.send(order["customer_email"], "Your order is confirmed!")

@subscribe("order.placed")
async def update_inventory(msg):
    for item in msg.data["items"]:
        await Product.get(item["id"]).reduce_stock(item["qty"])

# Wildcard matching — receives order.placed, order.cancelled, order.shipped, etc.
@subscribe("order.*")
async def log_all_orders(msg):
    print(f"Order event: {msg.topic}, data: {msg.data}")

Publishing

# Async — awaits delivery to subscribers
await publish("order.placed", data=order_dict, source="checkout")

# Sync — fire-and-forget, schedules async subscribers
publish_sync("order.placed", data=order_dict, source="checkout")

The Message Object

Subscribers receive a Message object with:

Attribute	Description
`msg.topic`	The topic string that was published
`msg.data`	The payload passed to `publish()`
`msg.source`	The source identifier (optional)

When to Use PubSub vs Direct Calls

Use PubSub for side effects: notifications, analytics, logging, cache invalidation. Use direct calls for anything that needs a return value or must complete before the response is sent.

class Order(Entity, table=True):
    customer_email: str
    total: float

    async def place(self, request=None):
        self.save()
        # Direct call — need the result before responding
        receipt = await PaymentService.charge(self.total)
        # PubSub — fire-and-forget side effects
        await publish("order.placed", data=self.model_dump(), source=str(self.id))
        return receipt

SSE Streaming

Client connects the PubSub system to the browser via Server-Sent Events. Clients subscribe to topics; when matching events are published, connected browsers receive updates instantly.

from nitro.events import Client
from nitro.events.starlette import emit_elements, emit_signals

# SSE endpoint
async def sse_stream(request):
    user_id = request.cookies.get("user_id", "anonymous")
    async with Client(topics=["updates.*"], source=user_id) as client:
        async for msg in client.stream():
            yield msg.data

Publishing UI updates from inside a handler:

from nitro.events.starlette import emit_elements

@post()
async def increment(self, request=None, amount: int = 1):
    self.count += amount
    self.save()
    user_id = request.cookies.get("user_id") if request else None
    emit_elements(
        ModelTable(Counter, id="counter-table"),
        topic="updates.ui",
        source=user_id,
    )
    return self.count

emit_elements serializes the HTML fragment and publishes it to the topic. Any Client subscribed to "updates.ui" (or "updates.*") will receive it and Datastar will patch the DOM.

Migration from v1 (Blinker-based)

The following APIs have been removed in v2:

on(), emit(), emit_async(), event() — replaced by the routing registry and PubSub
Event, Namespace, default_namespace — no longer needed
filter_signals(), ANY — use PubSub wildcard patterns instead

Key changes:

Routing and events are now separate. HTTP dispatch (entity methods) goes through the handler registry, not a signal bus.
Entity method handlers are auto-registered via @post()/@get() decorators — no manual wiring.
PubSub handler signatures changed. Handlers now receive a single msg: Message argument instead of (sender, **kwargs).
source= replaces sender= in emit_elements and emit_signals helpers.
publish_sync() replaces emit() for synchronous fire-and-forget publishing.