concresce 3.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. Optional tuning is available when you need it.

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())

Core Mechanics

• Event Loop Batching (default): With zero configuration, concresce yields exactly once (await asyncio.sleep(0)) to the event loop. Under heavy load, batches are massive. Under low load, batches execute instantly. The tuning emerges purely from the traffic itself.
• Native Type Routing: The leader does not need to call a special scatter function. If your database returns a list or tuple, the system unzips it by index. If it returns out-of-order or missing data, just return a dictionary ({id: result}) and concresce handles exact key-based routing automatically. 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 the heavy processing phase, the exception is intercepted and safely replicated to all suspended followers. Nobody hangs, and the stack unwinds naturally.

Optional Tuning

The default single-tick window is enough for most workloads. When you need more control, three opt-ins are available; none of them change the default behavior.

Time window & max size

By default a batch spans a single event-loop tick. Wrap calls in window(seconds) to keep pooling across ticks for a fixed time. Within a window, add max_size(n) to close the batch early once it reaches n items — max_size only has an effect alongside a window, since the default single-tick batch already closes immediately:

from concresce import batch, collect, window, max_size

@batch
async def enrich(item):
    return await bulk_lookup(await collect(item))

# Pool everything that arrives within 5 ms, or as soon as 100 items queue up.
with window(0.005), max_size(100):
    results = await asyncio.gather(*(enrich(i) for i in stream))

Partitioned batches

By default all concurrent calls to a decorated function share one batch. Pass key=fn to partition them so only calls whose key(*args, **kwargs) matches are pooled together — useful for keeping distinct instances, models, shards, or tenants from contaminating each other:

class Model:
    @batch(key=lambda self, prompt: id(self))
    async def infer(self, prompt):
        return await self.backend.infer_batch(await collect(prompt))

Stats

Every decorated function exposes a cumulative stats dict (batches, items, and the largest max batch seen):

@batch
async def fetch(item):
    return await bulk_fetch(await collect(item))

await asyncio.gather(fetch(1), fetch(2), fetch(3))
print(fetch.stats)  # {"batches": 1, "items": 3, "max": 3}