Extensions

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:

app_configs/app.ini ini
[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:

fluid/events/models.py python
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:

main.py python
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
 

Startup hooks get their own chapter later on; for now it is enough to know that this one runs once, inside the running loop, before the server accepts its first request. Handlers must accept exactly one argument: the event 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. Our add_model handler is the obvious place — only it changes here, get_model stays exactly as it was in the Mail chapter:

fluid/api/models.py python
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}
 

trigger is deliberately synchronous — there is nothing to await, because nobody is waiting. If you need an answer back, that is what queries are for, and they are next.

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:

browser html
<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:

fluid/events/models.py python
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.

 

Broadcasts use a bounded per-listener buffer sized by EVENTS_EVENT_QUEUE_SIZE. A consumer that falls behind loses its oldest events — you get a warning in the log naming the channel and the dropped payload, but the message is gone. Treat delivery as best-effort: for dashboards and live notifications it works nicely, but it is not a durable queue.

Continue reading

From here you can continue straight with Cache.