Interception

This SDK classifies every argument passed to .delay() or .apply_async() by its Python type, automatically, against a built-in and user-extensible registry. See Classifying arguments below.

This SDK runs enqueue interceptors — functions you register with queue.intercept() that inspect the whole call (task name + args) and decide its fate. See Enqueue interceptors below.

This SDK runs enqueue interceptors — functions you register with flexiq.intercept() that inspect the whole call (task name + payload) and decide its fate. See The interceptor pipeline below.

This SDK hooks into it nowhere. There is no intercept, no registry, and no producer-side hook of any kind — enqueue takes a TaskCall<T> and writes it.

Most of what interception is for does not arise, because the call is typed: <task>::call(..) mirrors the function's own signature, so a value the task does not accept is a compile error rather than something to catch at the producer. What is left — validating a value the type system allows, or rejecting one — is a function you call before call(..). See Validating before the call below.

Classifying arguments

Enable interception on the Queue constructor:

queue = Queue(db_path="tasks.db", interception="strict")

Without interception, values are passed directly to the serializer. A SQLAlchemy session or file handle would either raise a serialization error at enqueue time or produce a broken payload that fails on the worker.

Modes

"strict" is recommended for production — it surfaces problems at call time rather than causing silent task failures.

Classification strategies

Every argument gets one of five strategies:

Built-in CONVERT types

These are converted automatically when interception is enabled:

Built-in REDIRECT types

These connectors are automatically detected and replaced with a resource injection marker. The worker injects the named resource instead of attempting to deserialize a live connection object:

The resource name is the key you use in @queue.worker_resource("name"). If your resource has a different name, register a custom redirect with register_type().

REDIRECT is a type shift, not just a value swap: the caller passes a live Session (or Redis, MongoClient, ...) to .delay(), but the worker's task function receives whatever the named resource's factory returns — for "db", that's typically a sessionmaker you call to get a session, a different type than what the caller passed in. See Dependency Injection for how the factory return value is injected.

Built-in PROXY types

These objects are deconstructed to a recipe dict and rebuilt by the worker:

See Resource Proxies for security options and handler details.

Built-in REJECT types

These are always rejected because they cannot cross process or serialization boundaries:

• Thread synchronization primitives (Lock, RLock, Semaphore, Event)
• socket.socket
• Generator objects
• Coroutine objects
• subprocess.Popen
• asyncio.Task / asyncio.Future
• contextvars.Context
• Multiprocessing Lock and Queue

Each rejection includes a message explaining why and suggests alternatives.

Registering custom types

Add custom rules for types not covered by the built-ins:

from myapp import MyDBClient, MoneyAmount, APIConnection

# Treat a custom DB client as a worker resource (worker must have "my_db" registered)
queue.register_type(MyDBClient, "redirect", resource="my_db")

# Convert a custom value type to something serializable
queue.register_type(
    MoneyAmount,
    "convert",
    converter=lambda m: {"__type__": "money", "value": str(m.value), "currency": m.currency},
    type_key="money",
)

# Reject with a helpful message
queue.register_type(
    APIConnection,
    "reject",
    message="API connections are process-local. Register it as a worker resource instead.",
)

register_type() requires interception to be enabled ("strict" or "lenient"). Calling it when interception is "off" raises RuntimeError.

Constructor parameters

Analyzing arguments

Inspect how interception would classify arguments without actually transforming them:

from myapp.tasks import queue

report = queue.analyze_arguments(
    args=(user_session, "Hello"),
    kwargs={"attachment": open("file.pdf", "rb")},
)
print(report)
# Argument Analysis:
#   args[0]            (Session)              → REDIRECT (redirect to worker resource 'db')
#   args[1]            (str)                  → PASS
#   kwargs.attachment  (BufferedReader)        → PROXY (handler=file)

analyze_arguments() is a development and debugging tool. It reads the registry but makes no changes to arguments, and returns an empty report when the queue was created with interception="off". The per-task variant my_task.analyze(...) takes the task's own call arguments directly.

Interception metrics

stats = queue.interception_stats()
# {
#   "total_intercepts": 1200,
#   "total_duration_ms": 216.0,
#   "avg_duration_ms": 0.18,
#   "strategy_counts": {"pass": 800, "convert": 250, "redirect": 100, "proxy": 50, "reject": 0},
#   "max_depth_reached": 3,
# }

total_intercepts always equals the sum of strategy_counts.

See Observability for Prometheus metrics and dashboard endpoints.

Enqueue interceptors

queue.intercept(interceptor) registers a function that runs at the very start of every enqueue — before per-task defaults, middleware, and gates — and decides what happens to the call:

import { Interception } from "@byteveda/flexiq";

queue.intercept((taskName, args) => {
  if (taskName === "chargeCard" && (args[0] as number) < 0) {
    return Interception.reject("charge amount must be non-negative");
  }
  return Interception.pass();
});

An interceptor returns one of four outcomes:

Rejecting aborts the enqueue — the job is never created and the error propagates to the caller:

queue.enqueue("charge", [-5]); // throws — nothing is enqueued

Multiple interceptors chain in registration order, each seeing the previous one's (possibly redirected) task name and args. redirect is rejected for enqueueMany (a batch is stored under one task name) and for tasks registered with per-task codecs — the redirect target's codec chain can't be resolved from a bare name.

Interceptors vs. onEnqueue

queue.intercept() is distinct from the onEnqueue middleware hook: intercept runs first and can redirect to a different task or reject outright; onEnqueue runs after and mutates the (possibly redirected) context in place — use it to rewrite enqueue options (metadata, priority, and so on) rather than the payload. Most apps need only one.

Why no proxies?

Some SDKs also ship object proxies — a system for shipping non-serializable objects (file handles, sessions) through the queue. Node doesn't need it: a handler closes over its own resources, or pulls them from dependency injection. Keep non-serializable state on the worker and pass plain data through the queue.

The interceptor pipeline

flexiq.intercept(interceptor) registers an Interceptor — a functional interface that sees the task name and payload and returns one of four Interception strategies:

flexiq.intercept((taskName, payload) -> {
    if (payload instanceof String s && s.startsWith("pw:")) {
        return Interception.convert("***"); // redact before it reaches storage
    }
    return Interception.pass();
});

Interceptors run synchronously in registration order; each sees the previous one's result. A Redirect retargets the rest of the pipeline — later interceptors, middleware, and gates all see the new task name and payload.

Validation

Rejecting from an interceptor aborts the enqueue — the job is never created and the error propagates to the caller:

flexiq.intercept((taskName, payload) -> {
    if (taskName.equals("charge") && ((Order) payload).total() < 0) {
        return Interception.reject("charge amount must be non-negative");
    }
    return Interception.pass();
});

flexiq.enqueue(CHARGE, invalidOrder); // throws InterceptionException — nothing enqueued

Rewriting options — the onEnqueue hook

Interceptors decide the payload's fate; to rewrite enqueue options or attach metadata, use middleware onEnqueue. It receives a mutable EnqueueContext after interceptors have run — every enqueue runs interceptors → onEnqueue middleware → enqueue gates, then serializes, codec-encodes, and submits. Gates see the payload that will actually be stored, after any rewrites.

Edge cases

• Batches — enqueueMany runs each payload through the interceptors, so a batch can't bypass the contract. Convert rewrites the item; Reject fails the whole batch; Redirect is unsupported in a batch (it would move an item out of the single-task batch) and throws.
• Redirect + payload codecs — redirecting a task that has payload codecs throws: the source task's codec chain can't be applied to the target task without corrupting the payload.
• Null result — an interceptor returning null is a bug and throws InterceptionException.

Validating before the call

With no hook to register, the place to centralise a producer-side rule is a function that builds the call. Give it the validation and hand back a TaskCall, and every call site goes through it because there is nothing else to call:

use flexiq::TaskCall;

/// The only sanctioned way to enqueue a charge.
fn charge_call(amount_cents: i64) -> Result<TaskCall<charge_card>, ChargeError> {
    if amount_cents < 0 {
        return Err(ChargeError::Negative);
    }
    Ok(charge_card::call(amount_cents))
}

queue.enqueue(charge_call(-5)?)?;  // never reaches the queue

Redirecting is the same idea with a second arm — but note that the return type pins the task, so two task types need two functions or an enum. TaskCall<T> is generic in T, and charge_card and charge_card_v2 are different types.

What the encoder refuses on its own

Some values pass the type checker and still have no representation in the call envelope. Those are caught at enqueue, not at call(..), and come back as a QueueError:

task `charge_card` could not encode its arguments: cannot encode task argument:
a u64 out of range for the envelope's signed integer

The refusals are structural rather than configurable, because a value that encoded differently would still interoperate and quietly break auto: idempotency keys:

The second one catches more than it sounds like. Only three things make a legal key: a &str/String, a char, and a fieldless enum variant — which travels as its own name. A bool, any integer, a float, bytes, None, (), and every composite are refused, so a HashMap<i64, _> argument fails at enqueue even though it serializes fine everywhere else. Key by String instead.

Everything else encodes: f32/f64, byte strings, Option, Vec, structs, and both fieldless and data-carrying enum variants (the latter as serde's externally tagged one-entry map). The serializer runs in binary mode — is_human_readable() is false — so a Uuid round-trips as its bytes rather than its hyphenated string.

Redaction

There is no hook that could redact for you, so keep the secret out of the payload instead of trying to strip it later. Enqueue a key and let the task read the value from the store it came from:

// Not: charge_card::call(card_number)
queue.enqueue(charge_card::call(vault_token))?;

That is worth doing regardless of the runtime — a payload is at rest in storage, readable by anything with the connection string.

Struct field order matters

If an argument is a struct, its fields encode in declaration order and nothing sorts them. Reordering the fields changes the encoded bytes, which changes the auto: key .idempotent() derives — so a field moved during a refactor silently stops deduping against jobs enqueued by the old build.