concresce 0.2.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 constraint, collect() to suspend the execution and pool data, and scatter() to fan the results back out to the original callers.

There is zero configuration: the batch window is dynamically defined by the event loop's microtask queue, and routing is explicit and positional — caller i receives results[i].

import asyncio
from concresce import batch, collect, scatter

@batch
async def fetch_user_score(user_id: int) -> int:  # honest types: int -> int
    # 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 fans the results back out, one per caller, in collect() order.
    # scatter() returns this caller's own share, keeping the signature honest.
    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())

Because distribution happens through scatter() rather than through your return value, the decorator never inspects what you return — functions whose per-item results genuinely are lists or dicts route them untouched. If your bulk API returns results out of order or keyed, put them back in collect() order yourself before scattering:

@batch
async def enrich(ip: str) -> dict:
    ips = await collect(ip)
    by_ip = await geo.bulk_lookup(ips)     # returns {ip: {...}, ...}
    return scatter([by_ip[i] for i in ips])  # a missing key raises at YOUR line

Methods

@batch works directly on methods — no cached_property wrapper or per-instance bookkeeping required. Each instance gets its own batch, so concurrent calls on the same instance coalesce while calls on different instances never merge into one another's leader:

class ScoreClient:
    def __init__(self, tenant):
        self.tenant = tenant

    @batch
    async def fetch(self, user_id):
        batch_ids = await collect(user_id)
        scores = await self.db.bulk_fetch_scores(self.tenant, batch_ids)
        return scatter(scores)

a, b = ScoreClient("acme"), ScoreClient("globex")
# a's two calls share one query; b runs its own — no cross-tenant bleed.
await asyncio.gather(a.fetch(1), a.fetch(2), b.fetch(9))

Per-instance isolation is keyed by a weak reference to the receiver, so the instance must be weak-referenceable — the default. A class only opts out by declaring __slots__ without __weakref__.

Core Mechanics

• Event Loop Batching: concresce yields exactly once to the event loop, so a batch spans a single event-loop tick. Under heavy load, batches are large; under low load, they execute instantly.
• Positional Scatter: scatter() takes a list or tuple with exactly one result per pooled caller and routes it by position — anything else raises BatchRoutingError for every caller instead of hanging or silently handing out None. Your return value is never inspected, so per-item results that are themselves lists or dicts just work.
• Contextual Safety: If the leader returns without calling scatter(), concresce detects the structural failure and raises a RuntimeError for every caller rather than leaving followers deadlocked.
• 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. If the crash happens after scatter(), followers keep their already-delivered results and only the leader's caller sees the exception.
• Loop Isolation: Batch state is keyed by the running event loop, so the same decorated function used from multiple event loops (e.g. one per thread) never pools items across loops.