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
EVENTS_CONFIGURE_SOCKET = 1

EVENTS_CONFIGURE_SOCKET is on by default and is what mounts the socket and injects the client. Turn it off for an app that only ever uses events server-side — a worker, or an API without a browser in front of it — and everything below still works, minus the browser half.

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


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

    # A server-side reaction. 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}")

Note that this is a function, not module-level code. Declaring a channel starts a broadcast loop behind it, and a loop needs an event loop to live in. The framework is forgiving about the timing — a channel declared before the server is up is simply queued and wired in once the loop exists, so a signal at module level does work — but handing registration to a startup hook is still the shape to learn, for a reason that has nothing to do with timing: it puts every contract of a module in one place, where you can read them all at once. We come back to the one place where the timing genuinely bites at the end of this chapter.

def prepare_fluid() -> Fluid:
    app = Fluid(__name__)

    from fluid.app import app_router
    from fluid.api import api_router
    app.include_router(app_router)
    app.include_router(api_router)

    from fluid.events.models import register as register_events
    app.startup_hook(register_events)

    my_ext.expand_fluid(app)

    return app

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. Our add_model handler is the obvious place — only it changes here, get_model stays exactly as it was in the Mail chapter:

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

from fluid.models import MyModel
from fluid.services.notify import welcome


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)

    welcome("friend@example.org", model.value)

    # Publishing is fire-and-forget and does not block: it drops the
    # payload into the channel and returns. Every handler runs and every
    # subscribed browser gets it, on the loop behind that channel.
    events.trigger("model:created", {
        "id": model.id,
        "value": model.value
    })

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

Reaching the browser

The browser half rides in with the surface layer we'll set up in the Frontend chapters: it injects a small events.js client that lives under window.wf.ext.events. Once that is in place, subscribing to a public event from a page takes a handful of lines — 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>

The socket knows who is asking

Here is the part that makes the browser half more than a notification pipe. When a message arrives on /ws/events, the framework wraps the handling in the same request context an HTTP handler runs in — only the request is the WebSocket itself. Starlette treats both as the same kind of connection, so everything that reads "the request" keeps working: the session cookie is there, so the signed-in user resolves; the Accept-Language header is there, so _() answers in the visitor's language; url_for resolves.

Which means a query may look up its own caller instead of being told who they are:

from webfluid.core.context import FluidContext
from webfluid.core.ext import events, security


@events.query("my:overview", internal=False)
async def overview(_):
    ctx = FluidContext.try_current()
    request = ctx.request if ctx is not None else None
    if request is None: return None

    async for user in security.user_service.current_user_fn(request):
        if user is None: return None
        return {"greeting": f"Hello {user.username}"}

The same containment applies to failure. A query that raises answers {"error": "…"} and the socket stays open for the next message, and a message that is not a JSON object at all — a number, a stray binary frame — is answered with a named error rather than tearing the connection down for that visitor. Closing a tab is just a disconnect, and the subscriptions that belonged to it are dropped with it.

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


def register():
    events.create_signal("model:created", internal=False)

    @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}")

    @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")

Queries have no broadcast loop behind them, so they would survive being declared at import time — but keeping every contract of a module in one register() means you never have to remember which kind needs the loop and which does not.

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

Where registration goes wrong

We promised to come back to the timing. A channel's consumer loop is a task, and a task inherits the context it was started in. Declare an event from inside a request handler and that loop keeps the visitor's request for the lifetime of the process — so every handler on that channel from then on believes it is serving a request that ended long ago, and a handler that raises does so into a caller that no longer exists, leaving nothing in the log. Nothing about this fails loudly, which is exactly what makes it worth naming.

The other end of the same rope: an event registered after the server is running and from outside its loop has missed the moment when queued channels were wired up. That one does fail loudly — a FrameworkException saying so. Let it propagate. If you catch it, you are left with a name the framework will happily accept triggers for and no loop that will ever read them.

Continue reading

From here you can continue straight with Cache.