Jobs and Broadcasts

Background task processing with pluggable job providers, and real-time updates pushed to the browser over server-sent events.

Fymo is a synchronous, per-request framework built on WSGI. There’s no async runtime hiding underneath it.

Jobs and broadcasts are the two escape hatches for what doesn’t fit a request and response cycle. Jobs handle work that outlives the request. Broadcasts push updates to the browser that it never asked for.

Declaring a task

# app/jobs/emails.py
from fymo.jobs import task
from app.support.mailer import send

@task
def send_welcome_email(uid: str, email: str) -> None:
    send(to=email, template="welcome", uid=uid)

Every function you define at the top level of a job file becomes a task, as long as its name doesn’t start with an underscore. Fymo finds it through the same discovery that picks up remote functions and broadcast channels: it just walks the directory.

Adding @task doesn’t change what gets discovered. An undecorated function is still registered as a task, purely for backward compatibility. What changes is a warning: leave it off and fymo logs a note suggesting you add it.

That warning exists because job files are meant to stay thin. It’s easy to write a small helper, forget it’s public and top-level, and have it accidentally become submittable. Keep real logic in your support modules, and underscore-prefix any helper in a job file that isn’t meant to be a task.

Task names must be unique across every job module. If two modules both define send_welcome_email, fymo raises a startup error instead of silently overwriting one. Broadcast channel names follow the same rule.

Submitting a job

from fymo.jobs import get_job_provider
from fymo.remote import remote

@remote
def register(email: str, password: str) -> None:
    uid = create_account(email, password)
    get_job_provider().submit("send_welcome_email", uid=uid, email=email)

Calling submit() is fire-and-forget. It returns nothing, blocks nothing, and the request that called it moves on immediately.

Nothing tracks progress or result for you. If a task needs to report an outcome, it has to persist that itself, a database row is the usual choice. The rest of your app reads that outcome later, either through a poller or through the task’s own side effects.

Choosing a job provider

A job provider is a small interface with two jobs of its own. At startup, it wires every discovered task into whatever it uses internally. When you call submit(), it enqueues that call to run in the background. Two providers ship with fymo out of the box.

Provider	Where it runs	Durability	Needs
threaded (default)	In-process thread pool	Dies if the process restarts mid-run	nothing extra
procrastinate	A separate fymo jobs-worker process	Survives a restart, scales independently of the web tier	Postgres (DATABASE_URL), pip install fymo[procrastinate]

The threaded provider wraps fymo’s in-process job runner: a bounded thread pool with three workers by default. One bad task can’t take down the pool, or the request that submitted it.

It’s the fallback when jobs.provider is unset in fymo.yml. That mirrors the role password plays for auth, the default you get for free.

The procrastinate provider is a Postgres-native durable queue. A submitted job becomes a row, so it survives a restart of the web process. Whichever fymo jobs-worker process is listening picks it up, and you can run several of them.

You can reuse your app’s own database for this. Procrastinate keeps its state in its own tables, so it won’t collide with yours.

# fymo.yml
jobs:
  provider: procrastinate

Both providers log the same way: one line when a task starts, succeeds, or fails, with duration. Job arguments are never logged, only the task name, status, and timing, so sensitive input never leaks into your logs.

Running a worker

The threaded provider has no separate worker to run. By the time submit() returns, the task has already run.

Procrastinate does need one. Running fymo jobs-worker starts a process that opens its own database connection, registers every discovered task, and blocks, picking up jobs as they get deferred.

Calling run_worker() on a provider that isn’t a durable queue raises an error. The threaded provider simply has nothing for it to do.

fymo jobs-worker --dev sets FYMO_DEV=1 before anything else runs, which turns on .env loading. That matters here because DATABASE_URL usually lives in .env during development, not in fymo.yml itself.

Checking job status

fymo jobs-status

jobs-status asks the configured provider two questions: how many jobs are in each status, and what the most recent ones are. It’s read-only, meant for answering “is this job stuck” without hand-querying Postgres yourself.

Job status (procrastinate)
  failed     6
  succeeded  12

Recent jobs (newest first, up to 10)
  ID  TASK         STATUS   QUEUED AT
  18  boom         failed   2026-07-16 10:02:11
  ...

-n/--limit controls how many recent jobs print, 10 by default. --dev behaves exactly like jobs-worker --dev: it sets FYMO_DEV=1 so .env loading kicks in before the provider reads DATABASE_URL.

Not every provider can answer these questions. The threaded provider has no separate bookkeeping to read, since its jobs run and finish inside the same process that submitted them. A CLI invocation of jobs-status builds its own fresh provider a few milliseconds before asking it anything, and that fresh provider was never the web server that actually ran your jobs, so it has no way to see their outcomes. Rather than print confident zeros, it refuses:

✗ the 'threaded' job provider does not track job state — there is nothing
  to report. Providers backed by a durable queue (e.g. 'procrastinate')
  support `fymo jobs-status`; see docs/conventions.md for the app-level
  progress convention.

That’s an exit code of 1, not a crash. Procrastinate, backed by real Postgres tables, tracks state independently of any particular process and answers for real, as shown above.

job_counts() and list_recent_jobs() are optional methods on the job provider interface. The base implementation returns None from both, which means “this provider doesn’t track job state,” distinct from an empty dict or list, which would mean “tracked, and there’s nothing there yet.” A custom provider you wrote before this surface existed doesn’t need any changes: it already returns None for both by inheriting the base defaults, and jobs-status reports it as untracked rather than raising.

Declaring a channel

# app/broadcasts/posts.py
from typing import Literal, NotRequired, TypedDict
from app.data.db import get_db
from app.remote.posts import Comment, ReactionCounts
from fymo.remote import NotFound


class PostActivity(TypedDict):
    kind: Literal["comment_added", "reaction_updated"]
    comment: NotRequired[Comment]
    reactions: NotRequired[ReactionCounts]


def post_activity(slug: str) -> PostActivity:
    # Deny subscribing to a post that isn't there rather than silently
    # opening a channel nothing will ever publish to.
    if not get_db().fetchone("SELECT 1 FROM posts WHERE slug = ?", [slug]):
        raise NotFound(f"post '{slug}' not found")

A channel is just one function in a broadcasts file. It’s discovered the same way tasks are: non-underscore top-level functions only, with names unique across modules.

Three things come off that single function. Its signature defines the arguments a client passes when it subscribes. Its return annotation defines the payload type, and in dev mode fymo checks that against what you actually publish. Its body is the authorization guard.

That guard runs on every subscribe attempt, inside the same request scope remote functions get. So current_uid() and anything built on it works normally inside it. Return exactly False, or raise, and fymo rejects the subscription. Anything else, including a bare ... body, allows it.

Publishing events

def _publish_activity(slug: str, kind: str, **payload) -> None:
    # A broadcast failure must never fail the mutation that triggered it.
    try:
        from fymo.broadcast import publish
        publish("post_activity", slug=slug, data={"kind": kind, **payload})
    except Exception:
        logger.warning("post_activity broadcast failed for %s", slug, exc_info=True)

Publishing takes two different kinds of arguments on purpose. The keyword arguments you pass select which subscribers receive the event, and they have to match the channel function’s own signature. Passing slug="my-post" only reaches subscribers who opened that exact channel with that exact slug.

The data argument is different: it’s what gets sent, JSON-encoded straight onto the wire as an SSE frame.

Publishing is fire-and-forget, just like job submission. No subscribers means the payload is simply dropped, not an error. That’s why the example above wraps the call in a try/except block, so a broadcast hiccup never fails the comment it’s reporting on.

In dev mode, publishing checks your data against the channel’s declared return type. It logs a warning if a required key is missing, or if an unexpected one shows up.

This never blocks delivery. It’s a development-time nudge, not a runtime contract.

Subscribing from Svelte

<script>
  import { subscribe } from '$broadcast/posts';
  import { onDestroy } from 'svelte';

  let unsubscribe = subscribe.post_activity({ slug: post.slug }, (data) => {
    if (data.kind === 'comment_added') comments = [data.comment, ...comments];
    if (data.kind === 'reaction_updated') reactions = data.reactions;
  });

  onDestroy(() => unsubscribe());
</script>

Every broadcasts module gets its own generated file pair, mirroring how the $remote codegen works for remote functions. You import from $broadcast/<module>.

The generated types come straight from the channel function. Its parameters become the typed args object, and its return annotation becomes the payload type passed to your callback.

// generated: $broadcast/posts.d.ts
export interface PostActivity {
  kind: "comment_added" | "reaction_updated";
  comment?: Comment;
  reactions?: ReactionCounts;
}
export const subscribe: {
  post_activity(args: { slug: string }, onEvent: (data: PostActivity) => void): () => void;
};

Under the hood, subscribe.post_activity(...) opens a standard EventSource connection and parses each frame as JSON.

It connects to a URL shaped like /_fymo/broadcast/posts/post_activity?slug=..., generated for you automatically.

That reconnect-on-drop behavior comes from the browser’s EventSource, not from anything fymo adds. A subscription the server actively rejects, say the guard returned False, or the channel doesn’t exist, closes for good instead. That matches the fire-and-forget contract on the publish side: the subscription is just over.

How a subscription resolves

Here’s what happens when a subscription comes in. The server resolves <module>/<channel> from the discovery registry, then binds the query string against the channel function’s signature. A mismatch there is a 422, not a crash.

Next it runs the function body as the guard. A rejection there is a 403. Finally it hashes the module, channel name, and bound arguments together into a channel key.

That key is what the broadcast provider actually listens on. Two subscribers calling post_activity with different slug values never see each other’s events, even though they hit the same channel function, they’re just on different keys.

Each open SSE connection holds a thread for its lifetime. That’s fine on the threaded dev server and on gunicorn’s gthread worker class, but it can starve a sync worker pool under real subscriber load.

Broadcast transport

The default broadcast provider runs on Postgres, using the same LISTEN/NOTIFY primitive procrastinate uses for near-instant job pickup. It needs nothing beyond the Postgres database your fymo app already has.

How the Postgres transport works

It works across separate OS processes. A fymo jobs-worker process sends the notify. A web worker holding a matching listen picks it up. That’s how a background job ends up updating a page in real time.

Postgres also caps a NOTIFY payload at 8000 bytes. The provider enforces that loudly at publish time instead of truncating silently. Keep broadcasts small: publish an id and let the subscriber fetch the rest, not the whole blob.