Metadata-Version: 2.3
Name: concresce
Version: 3.0.0
Summary: Transparent temporal coalescing and inline micro-batching for Python.
Keywords: asyncio,batching,micro-batch,performance,coalesce
Author: Wannes Vantorre
Author-email: Wannes Vantorre <vantorrewannes@gmail.com>
License: MIT
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python :: 3.9
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Programming Language :: Python :: 3.14
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Classifier: Framework :: AsyncIO
Requires-Python: >=3.14
Project-URL: Homepage, https://codeberg.org/ProductionCode/concresce
Project-URL: Repository, https://codeberg.org/ProductionCode/concresce.git
Project-URL: Issues, https://codeberg.org/ProductionCode/concresce/issues
Description-Content-Type: text/markdown

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

```bash
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](#optional-tuning) is available when you need it.

```python
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:

```python
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:

```python
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):

```python
@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}
```
