Events

So far our app reacts to requests. The Events battery lets it react to things happening inside the app instead — a small publish / subscribe layer with a twist: the same events reach across to the browser through a managed WebSocket, so you can push live updates without writing any socket plumbing yourself.

Enabling events

Set EXT_EVENTS = 1. On startup the manager mounts a WebSocket at /ws/events and injects an events.js client into your pages. The depth of the internal queues is the only knob, defaulting to five:

[general]
SECRET_KEY = supersecret

[extensions]
EXT_SQLALCHEMY = 1
EXT_EVENTS = 1

[events]
EVENTS_EVENT_QUEUE_SIZE = 5

Signals, events and visibility

An event is identified by a name. Before you can trigger one, it has to exist. The simplest way to declare a pure broadcast channel is create_signal. Every event carries two flags worth understanding:

• singleton — whether the name may only ever be registered once.
• internal — whether the event stays server-only. Internal events are invisible to the browser; only public events (internal=False) can be subscribed to or triggered from the client.

Now that our app is structured around an app factory, event handlers deserve their own home. We'll keep them in a fluid/events package and declare a public signal that fires whenever a model is created:

from webfluid.core.ext import events


# A public broadcast channel the browser is allowed to listen to.
events.create_signal("model:created", internal=False)


# A server-side reaction. internal handlers receive the event data
# and run inside a fresh framework context for you.
@events.event("model:created", internal=False)
async def on_model_created(data):
    from webfluid.utils.logging import factory as log
    log.log(f"A new model appeared: {data}")

Triggering an event

From anywhere in your app you publish data to a channel with events.trigger. Every registered handler runs, and every subscribed browser receives the payload:

from webfluid.core.ext import db, events
from fastapi import Request
from fastapi.exceptions import HTTPException

from fluid.models import MyModel


async def add_model(request: Request):
    data = await request.json()
    value = data.get("value")
    if not value:
        raise HTTPException(status_code=400, detail="Bad Request")

    async with db.async_executor(model=MyModel) as e:
        model = await e.insert(MyModel(value), flush=True)

    await events.trigger("model:created", {
        "id": model.id,
        "value": model.value
    })

    return {"model_id": model.id, "value": model.value}

Reaching the browser

On the client the injected manager lives under window.wf.ext.events. You subscribe to a public event and register one or more handlers. The client transparently reconnects and keeps a listen loop alive for you:

<script type="module">
    const events = new window.wf.ext.events.EventManager()

    await events.subscribe("model:created")
    events.registerHandler("model:created", (data) => {
        console.log("A model was created:", data)
    })
</script>

Queries: asking for an answer

Events are fire-and-forget. When you need a value back instead, register a query and call events.request. A singleton query returns the single handler's result; a non-singleton one collects a list from all handlers:

from webfluid.core.ext import db, events
from sqlalchemy import select, func

from fluid.models import MyModel


@events.query("model:count")
async def count_models(_):
    async with db.async_executor(model=MyModel) as e:
        result = await e.exec(select(func.count(MyModel.id)), scalars=False)
        return result.scalar()


# Somewhere else in your app:
# total = await events.request("model:count")

On the server you can also consume a channel as an async stream with events.listen(name), which is handy for long-running consumers.

Continue reading

From here you can continue straight with Cache.