concresce 0.1.1

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.

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.

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

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)
        return await self.db.bulk_fetch_scores(self.tenant, batch_ids)

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.
• 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.