concresce 0.1.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 three primitives: @batch to define the barrier constraints, collect() to suspend the execution and pool data, and scatter() to distribute the results back to the original callers.

import asyncio
from concresce import batch, collect, scatter

@batch(max_window=0.05, max_size=100)
async def fetch_user_score(user_id):
    # 1. Execution pauses here.
    # Concurrent calls to this function pool their `user_id`s together.
    batch_ids = await collect(user_id)

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

    # 3. The leader distributes the results back to the suspended followers.
    # Every caller gets their specific integer back.
    return scatter(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

• Dual-Trigger Barrier: The suspension breaks when either max_size is reached (volumetric) or max_window expires (temporal). High throughput processes instantly; low throughput falls back to the maximum allowed latency.
• Keyed Routing: If your bulk API returns data out-of-order or drops missing items, you can pass a dictionary to scatter({id: result}). It will automatically route the correct result back to the specific caller based on their original input.
• Fault Propagation: If the leader crashes during the heavy processing phase, the exception is safely replicated to all suspended followers. Nobody hangs, and the stack unwinds naturally.
• Contextual Safety: If the leader attempts to return without calling scatter(), concresce detects the structural failure and safely aborts the followers with a RuntimeError rather than leaving them deadlocked.