Batching

Producer-side task batching primitives. Import paths:

from flexiq.batching import BatchAccumulator, BatchConfig, BatchedJobResult

For the conceptual overview and trade-offs, see Task batching.

@queue.task(batch=…)

Enable producer-side batching for a task. Accepted values:

@queue.task(batch={"max_size": 50, "max_wait_ms": 200})
def send_emails(items: list[Email]) -> None:
    for item in items:
        smtp.send(item)

The task function must accept a list as its first positional argument. Each .delay(item) call adds one element to the in-memory buffer; when the buffer reaches max_size or max_wait_ms elapses, flexiq enqueues one job whose payload is the full list.

Restrictions

BatchConfig

Frozen dataclass:

@dataclass(frozen=True)
class BatchConfig:
    max_size: int = 100
    max_wait_ms: int = 500

BatchConfig.normalize(value)

Class method used by the decorator. Accepts True / dict / BatchConfig / None / False and returns a BatchConfig or None. Useful in user code that builds dynamic batch configs.

cfg = BatchConfig.normalize({"max_size": 50})    # → BatchConfig(50, 500)
cfg = BatchConfig.normalize(True)                # → BatchConfig(100, 500)
cfg = BatchConfig.normalize(None)                # → None (disabled)

BatchedJobResult

Sentinel returned by Queue.enqueue() / task.delay() for batched tasks — there is no real job id until the batch flushes.

result = send_emails.delay({"to": "a@x.com"})
assert isinstance(result, BatchedJobResult)
assert result.id is None

If you need per-item results, batching is the wrong primitive — use canvas.chunks() instead, which creates one job per chunk and supports per-chunk retry.

Concurrency, rate limit, and retries

A batched flush is one job. That means:

• max_concurrent=N on a batched task allows N simultaneous batches, not N items.
• A rate limit of "10/m" allows ten flushes per minute (potentially thousands of items per minute).
• If the function raises, the entire batch retries — every item is reprocessed. Wrap items individually if you need per-item resilience.

Queue lifecycle

Queue.close() flushes any items still in the accumulator and shuts down the daemon flusher thread. An atexit hook does the same on process exit. Items in the buffer at hard crash (no atexit) are lost — this is the documented trade-off and matches Celery's Batches extension.

queue.close()  # blocks until pending batches flush

For tests that want to observe the loss path deterministically:

queue._batch_accumulator.shutdown(flush=False)
queue._batch_accumulator = None