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.

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.