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 its broadcast loop, and a loop needs an event loop to live in — which does not exist yet while your factory is assembling the app. So we hand the registration to a startup hook, and the framework calls it at the right moment:

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>

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.

Continue reading

From here you can continue straight with Cache.