Python/Guides/Extensibility/

Middleware

Cross-cutting hooks around enqueue, task execution, and job outcomes — logging, metrics, tracing, and enqueue-time rewrites.

Middleware wraps task execution and reacts to job outcomes — the place for logging, metrics, tracing, and enqueue-time validation that would otherwise be copy-pasted into every task. Register it once and it applies across every task, or a specific oneevery taskevery task; multiple middlewares compose in the order you register them.

Registering middleware

python

from flexiq import Queue, TaskMiddleware

class LoggingMiddleware(TaskMiddleware):
    def before(self, ctx):
        print(f"[START] {ctx.task_name} (job {ctx.id})")

    def after(self, ctx, result, error):
        status = "OK" if error is None else f"FAILED: {error}"
        print(f"[END] {ctx.task_name}: {status}")

queue = Queue(middleware=[LoggingMiddleware()])

queue.use({
  before: (ctx) => log.info("start", ctx.taskName),
  after: (ctx, result) => log.info("ok", ctx.taskName),
  onError: (ctx, err) => log.error("threw", ctx.taskName, err),
  onRetry: (e) => metrics.inc("retry", e.taskName),
  onDeadLetter: (e) => alertOps(e),
});

java

flexiq.use(new Middleware() {
    @Override
    public void before(TaskContext context) {
        log.info("start " + context.taskName);
    }

    @Override
    public void onError(TaskContext context, Throwable error) {
        log.error("threw " + context.taskName, error);
    }

    @Override
    public void onDeadLetter(OutcomeEvent event) {
        alertOps(event);
    }
});

Subclass TaskMiddleware and override only the hooks you need — unimplemented hooks default to a no-op on the base class.

Middleware is a plain object whose hooks are all optional — implement only what you need.

Middleware is an interface whose hooks are all no-op default methods — override only what you need.

Hooks

TaskMiddleware exposes 7 hooks:

Hook	Called when
`before(ctx)`	Before task execution
`after(ctx, result, error)`	After task execution (success or failure)
`on_retry(ctx, error, retry_count)`	A job fails and will be retried
`on_enqueue(task_name, args, kwargs, options)`	A job is about to be enqueued
`on_dead_letter(ctx, error)`	A job exhausts all retries and moves to the dead letter queue
`on_timeout(ctx)`	A job hits its hard timeout
`on_cancel(ctx)`	A job is cancelled during execution

ctx is a JobContext — the same object as current_job — exposing ctx.id, ctx.task_name, ctx.retry_count, and ctx.queue_name.

Lifecycle hooks dispatched from Rust

on_retry, on_dead_letter, on_timeout, and on_cancel are called by the Rust result handler after the scheduler records the outcome, firing after after() and after the corresponding event is emitted on the event bus. Exceptions raised inside these hooks are logged and do not affect job processing.

on_timeout fires when the maintenance reaper detects a stale job that exceeded its hard timeout — before on_retry (if it will be retried) or on_dead_letter (if retries are exhausted), so you can react to the timeout itself independently of the job's eventual fate.

Middleware exposes 8 hooks:

Hook	When	Awaited
`onEnqueue(ctx)`	Producer-side, inside `enqueue`, before serialization.	no (sync)
`before(ctx)`	Before each execution attempt.	yes
`after(ctx, result)`	After a successful attempt.	yes
`onError(ctx, err)`	When an attempt throws (before the retry/dead decision).	yes
`onCompleted(e)`	After a job completes successfully.	no
`onRetry(e)`	After the core schedules a retry.	no
`onDeadLetter(e)`	After a job dead-letters.	no
`onCancel(e)`	After a job is cancelled.	no

The execution hooks (before/after/onError) receive a TaskContext with taskName, jobId, and args. The outcome hooks receive an OutcomeEvent with taskName, jobId, queue, retryCount, timedOut (which separates a timeout from other failures), and durationMs (absent when nothing measured the run).

Middleware exposes 8 hooks:

Hook	When
`onEnqueue(context)`	Producer-side, inside `enqueue`, before serialization.
`before(context)`	Before each execution attempt.
`after(context, result)`	After a successful attempt.
`onError(context, error)`	When an attempt throws (before the retry/dead decision).
`onCompleted(event)`	After a job completes successfully.
`onRetry(event)`	After the core schedules a retry.
`onDeadLetter(event)`	After a job dead-letters.
`onCancel(event)`	After a job is cancelled.

The execution hooks receive a TaskContext with taskName, jobId, a mutable per-execution attributes() map shared across a job's hooks, job() with lazily-loaded metadata, and elapsedMs() — time spent on this execution so far, so after/onError needn't start their own timer. The outcome hooks receive an OutcomeEvent with taskName, jobId, error, retryCount, timedOut (which separates a timeout from other failures), and durationMs() (null when nothing measured the run).

Rewriting an enqueue

on_enqueueonEnqueueonEnqueue is unique among the hooks: it fires before the job is written to storage, and the context it receives is mutable — use it to validate, redact, or reshape what actually gets enqueued.

python

class PriorityBoostMiddleware(TaskMiddleware):
    def on_enqueue(self, task_name, args, kwargs, options):
        # Bump priority for urgent tasks during business hours
        if task_name.startswith("alerts."):
            options["priority"] = max(options.get("priority", 0), 50)

queue.use({
  onEnqueue: (ctx) => {
    // Bump priority for urgent tasks
    if (ctx.taskName.startsWith("alerts.")) {
      ctx.options.priority = 50;
    }
  },
});

java

flexiq.use(new Middleware() {
    @Override
    public void onEnqueue(EnqueueContext context) {
        // Bump priority for urgent tasks
        if (context.taskName.startsWith("alerts.")) {
            context.options(context.options().toBuilder().priority(50).build());
        }
    }
});

Keys present in options: priority, delay, queue, max_retries, timeout, unique_key, metadata.

EnqueueContext also exposes mutable args, and throwing from onEnqueue aborts the enqueue. This is the same producer-side seam as enqueue interceptors, which run first and can convert, redirect, or reject the call outright before onEnqueue sees it.

EnqueueContext also exposes payload()/payload(...) to replace the job's payload outright, and a mutable metadata() map that travels with the job (readable at execution via context.job().metadata()). Throwing from onEnqueue aborts the enqueue. For typed convert/redirect/reject decisions, use an interceptor instead — interceptors run before middleware onEnqueue.

Execution order

Global middleware (registered via Queue(middleware=[...])) runs first
Per-task middleware (via @queue.task(middleware=[...])) runs second
Within each group, middleware runs in registration order
after() and the Rust-dispatched outcome hooks run in that same forward order, not reversed — after() only fires for middleware whose before() didn't raise.

Every hook — before, after, onError, and the outcome hooks — runs in the same registration order for every middleware; the Node SDK doesn't reverse after the way an "onion" middleware model would.

Multiple middlewares run in registration order for every phase — before, after, onError, and the outcome hooks are not reversed.

Exception handling

If a middleware hook raises an exception:

before(): the exception is logged, but subsequent middleware before() hooks still run. The task executes normally.
after(): the exception is logged. Other after() hooks still run.
on_retry() / on_dead_letter() / on_timeout() / on_cancel(): logged and swallowed — these are notification hooks, not control flow.

Middleware exceptions never prevent task execution or result handling.

The execution hooks (before/after/onError) are awaited in the hot path: a rejected before or after fails the attempt like a thrown handler error. onError itself is wrapped so a throwing onError never masks the original task failure. The outcome hooks (onCompleted/onRetry/onDeadLetter/ onCancel) are isolated per middleware — one that throws or rejects is logged at debug and doesn't stop the rest from running.

The execution hooks run inside the attempt — a before or after that throws fails the attempt like a handler error, and is caught by the same onError path. Outcome hooks are isolated — one that throws is caught and logged so it never starves the others.

Per-task middleware

Apply middleware to a specific task using the middleware parameter on @queue.task:

python

@queue.task(middleware=[MetricsMiddleware()])
def process(data):
    ...

Per-task middleware runs after global middleware, in registration order.

There's no separate per-task registration — every middleware passed to queue.use() applies to every task. Scope a middleware's own hooks to specific tasks by checking ctx.taskName inside them, or by giving the middleware its own task-name filter (as the OpenTelemetry integration does with taskFilter).

There's no separate per-task registration — every middleware passed to flexiq.use() applies to every task. Scope a middleware's own hooks to specific tasks by checking context.taskName inside them.

Middleware vs hooks

Coming from Celery signals? Use these instead of connecting to task_prerun / task_postrun / task_failure: queue-level hooks for global behavior, or per-task middleware for scoped behavior.

flexiq has two systems for running code around tasks:

	Hooks (`@queue.on_failure`, etc.)	Middleware (`TaskMiddleware`)
Scope	Queue-level only	Queue-level or per-task
Interface	Decorated functions	Class with up to 7 hooks
Context	Receives `task_name, args, kwargs`	Receives `JobContext`
Enqueue hook	No	Yes (`on_enqueue`, can mutate options)
Retry hook	No	Yes (`on_retry`)
DLQ / timeout / cancel hooks	No	Yes
Execution order	After middleware	Before hooks

Middleware runs inside the task wrapper (closer to the task function), while hooks run outside. In practice, middleware before() fires first, then before_task hooks. On completion, on_success/on_failure hooks fire, then middleware after().

Combining with observability

Middleware is how the built-in observability integrations attach to task execution — compose them with your own middleware freely.

python

from flexiq import Queue
from flexiq.contrib.otel import OpenTelemetryMiddleware

queue = Queue(middleware=[
    OpenTelemetryMiddleware(),
    LoggingMiddleware(),
])

import { otelMiddleware } from "@byteveda/flexiq/contrib/otel";

queue.use(otelMiddleware());
queue.use({ before: (ctx) => log.info("start", ctx.taskName) }); // your own middleware

java

import org.byteveda.flexiq.contrib.FlexiQObservation;

flexiq.use(new FlexiQObservation(registry));
flexiq.use(new Middleware() { // your own middleware
    @Override
    public void before(TaskContext context) {
        log.info("start " + context.taskName);
    }
});

See the OpenTelemetry guide for setup details.

See the OpenTelemetry and Sentry integration guides for setup details.

See the Micrometer and Sentry integration guides for setup details.