concresce 4.0.0

Transparent temporal coalescing and inline micro-batching for Python.

concresce 💧

Transparent temporal coalescing and inline micro-batching.

Standard solutions to N+1 database or network bottlenecks require spinning up external message queues, background workers, or complex task graphs. concresce solves this dynamically. You write the function as if it processes a single item, and the runtime transparently merges concurrent calls into a single batch in the background.

uv add concresce

The Difference

Concept	Standard Async Loop	Message Queues (Celery/Kafka)	concresce
Network Footprint	N requests for N items.	1 request for N items.	1 request for N items.
Architectural Overhead	None.	High (Requires external broker).	None (Pure inline code).
Return Routing	Native variables.	Complex (Webhooks / Polling).	Native variables (Futures resolve).

Usage

You need exactly two primitives: @batch to define the barrier constraint, and collect() to suspend the execution and pool data.

By default there is zero configuration: the batch window is dynamically defined by the event loop's microtask queue, and routing is handled natively by your return types. A single optional window knob is available when you need a wider collection window.

import asyncio
from concresce import batch, collect

@batch
async def fetch_user_score(user_id):
    # 1. Execution pauses here.
    # Concurrent calls inside the current event loop tick pool their `user_id`s.
    batch_ids = await collect(user_id)

    # 2. Only ONE execution path (the leader) resumes from this point.
    # The others remain safely suspended via Exception-driven control flow.
    print(f"Making 1 network call for {len(batch_ids)} users...")
    bulk_results = await db.bulk_fetch_scores(batch_ids)

    # 3. The leader returns the raw bulk list or dictionary.
    # The decorator natively maps and distributes the results back to the followers.
    return bulk_results


async def main():
    # Fire off 5 requests simultaneously
    results = await asyncio.gather(
        fetch_user_score(1),
        fetch_user_score(2),
        fetch_user_score(3),
        fetch_user_score(4),
        fetch_user_score(5)
    )

    # Returns:[100, 250, 190, 300, 120]
    print(results)

asyncio.run(main())

The window parameter

By default a batch spans a single event-loop tick: the leader yields once (await asyncio.sleep(0)) and then processes whatever pooled during that tick. That is ideal under load, but on bursty traffic the items you want to coalesce can land a few milliseconds apart, in separate ticks.

Pass a window to @batch to widen the collection window. It is a datetime.timedelta that decides how long the leader sleeps before it collects and runs the batch — every call that arrives during that window joins the same batch.

from datetime import timedelta
from concresce import batch, collect

@batch(window=timedelta(milliseconds=5))
async def fetch_user_score(user_id):
    batch_ids = await collect(user_id)
    return await db.bulk_fetch_scores(batch_ids)

• @batch (bare) — leader sleeps 0; the batch is one event-loop tick. This is the default.
• @batch(window=timedelta(milliseconds=5)) — leader sleeps 5 ms; everything that arrives in that window is coalesced.

Widening the window trades a little latency for larger, more efficient batches.

Core Mechanics

• Event Loop Batching: With the default zero-length window (timedelta(0)) concresce yields exactly once to the event loop. Under heavy load, batches are large; under low load, they execute instantly. A wider window simply extends how long the leader waits before collecting.
• Native Type Routing: The leader does not call a special scatter function. If it returns a list or tuple, the system unzips it by index. If it returns out-of-order or missing data, return a dictionary ({id: result}) and concresce routes by key. Any other return type — or a sequence whose length does not match the number of callers — raises BatchRoutingError for every caller instead of hanging.
• Fault Propagation: If the leader crashes during processing, the exception is intercepted and replicated to all suspended followers. Nobody hangs, and the stack unwinds naturally.