Defining jobs

A job handler is a plain function registered with the @worker.job(...) decorator. The decorator returns the function unchanged, so your handlers stay ordinary, unit-testable functions.

from zeridion_flare.worker import FlareWorker, JobContext

worker = FlareWorker()

@worker.job("SendWelcomeEmail", queue="email", max_attempts=5, timeout=60)
def send_welcome(payload, ctx: JobContext) -> None:
    send_email(payload["email"])
    ctx.report_progress(1.0)

A payload handler takes two positional parameters: the decoded payload and the JobContext. The first is the job's data; the second carries metadata, progress reporting, and the cancellation handle.

The job type is explicit

The first argument to @worker.job(...) is the job type — the routing key that ties an enqueued job to this handler. It is always an explicit string (there is no name derivation), so a worker reliably matches jobs enqueued under the same string from any language or service sharing the queue.

Registering the same job type twice raises a ValueError at startup.

Per-job configuration

Argument	Default	Meaning
queue	"default"	The queue this job type is served from.
max_attempts	3	Advertised attempt ceiling for this job type.
timeout	1800 (30 min)	Per-job timeout in seconds.

The server is the source of truth for retry and dead-letter decisions; these values describe the worker's defaults for the job type.

Typed payloads

By default the payload is the decoded JSON value — a dict for an object body. Annotate the first parameter and the worker hydrates it for you:

• Dataclass — fields are populated from the JSON object (unknown keys are ignored; nested dataclasses hydrate recursively).

  from dataclasses import dataclass
  
  @dataclass
  class NewUserEvent:
      email: str
      name: str
  
  @worker.job("SendWelcomeEmail", queue="email")
  def send_welcome(payload: NewUserEvent, ctx) -> None:
      print(payload.email, payload.name)

• Pydantic v2 model — validated via the model's own parsing. This requires the optional extra:

  pip install "zeridion-flare[pydantic]"

  from pydantic import BaseModel
  
  class NewUserEvent(BaseModel):
      email: str
      name: str
  
  @worker.job("SendWelcomeEmail")
  def send_welcome(payload: NewUserEvent, ctx) -> None:
      ...

  Pydantic is never imported unless it is installed — the worker detects it at runtime.

• No annotation / dict — the raw decoded dict is passed through. A null payload arrives as None.

If payload parsing raises, the job is acknowledged as failed and the server applies its retry policy.

Handling failures

Returning normally acknowledges success. Raising any exception acknowledges failure — the worker captures the exception type, message, and traceback and sends them with the failure so they show up against the job. You do not decide retry-vs-give-up in the handler; the server does, based on the attempt count.

A poison payload (one that can never succeed) should raise a clear, recognizable exception so it is easy to spot in the dashboard; the server governs how many times it is retried before it is dead-lettered.

See also

• Recurring jobs
• Job context
• Configuration