Metadata-Version: 2.4
Name: aiofence
Version: 0.0.2
Summary: Python asyncio-native cancellation context library
Project-URL: Repository, https://github.com/stanislaushimovolos/aiofence
Author: Shimovolos Stanislav
License-Expression: MIT
License-File: LICENSE
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Programming Language :: Python :: 3.14
Requires-Python: <3.15,>=3.12
Description-Content-Type: text/markdown

<p align="center">
  <img src="docs/images/logo.png" alt="aiofence" />
</p>

# aiofence

[![codecov](https://codecov.io/gh/stanislaushimovolos/aiofence/branch/main/graph/badge.svg)](https://codecov.io/gh/stanislaushimovolos/aiofence)

Multi-reason cancellation contexts for Python asyncio. Inspired by Go's `context.Context`, aiofence provides a cancellation context that propagates hierarchically through your application via `ContextVar` — no need to thread events, flags, or tokens through every call signature. Declare cancellation sources once at the boundary — inner code just wraps cancellable work in a context manager and doesn't care about the actual reasons, though it can inspect them if needed.

## Motivation

asyncio can cancel tasks mechanically — `task.cancel()`, `asyncio.timeout()` — but it can't tell you *why* you were cancelled or let you control *when*. Your task is abruptly killed at the next `await`, whether it's ready or not. When multiple cancellation sources exist (timeout, client disconnect, graceful shutdown), user code is forced to propagate events and flags through every call signature, spawn background listeners that watch for signals and cancel your task, and shield cleanup code so that a second cancel request doesn't kill the task mid-cleanup. Without a single centralized object that owns all of this, it gets messy fast:

```python
async def handle_request(request, shutdown_event, timeout=30):
    try:
        async with asyncio.timeout(timeout):
            while not shutdown_event.is_set():
                chunk = await get_next_chunk()
                if request.is_disconnected():
                    break
                await process(chunk)
    except TimeoutError:
        ...
    except asyncio.CancelledError:
        # shutdown? disconnect? something else?
        ...
```
For a deeper dive into the problem and design rationale, see [this Medium post](https://medium.com/p/8cdf8c5d519e).

`aiofence` solves this. Declare all cancellation sources once, composably. The callee doesn't even know cancellation exists:

```python
with Fence(TimeoutTrigger(30), EventTrigger(shutdown, code="shutdown")) as fence:
    result = await fetch_and_transform()

if not fence.cancelled:
    await save(result)
else:
    print(fence.reasons)       # (CancelReason(message='timed out after 30s', ...),)
    print(fence.cancelled_by("shutdown"))  # True / False
```

### What about `asyncio.shield()`?

`shield()` prevents cancellation from reaching shielded code, but it works from the opposite direction — you protect everything that *must not* be cancelled. In practice this means wrapping database writes, state transitions, logging, and cleanup individually, and each function needs to know whether it's cancel-safe.

aiofence comes at it differently: most code doesn't know cancellation exists. You only wrap the expensive, safely-interruptible parts — the operations you *want* to cancel. For example, in an LLM inference service, you don't want to cancel database queries or response formatting. You want to cancel the LLM call that's burning GPU time for a client that already disconnected:

```python
with Fence(EventTrigger(client_disconnect), TimeoutTrigger(budget)) as fence:
    result = await llm.generate(prompt)  # cancellable

await db.save(result or fallback)  # always runs, no shield needed
```

### Why not anyio?

anyio is one of the best async libraries in the Python ecosystem, and its `CancelScope` is a more powerful and general cancellation model than what asyncio provides natively. aiofence is narrower in scope and makes different trade-offs:

1. **Drop-in for existing asyncio code.** anyio introduces its own cancellation semantics (`CancelScope`, level-triggered cancellation, nested scope trees). If your app is already built on pure asyncio, adopting anyio's model is a significant migration. aiofence works directly with asyncio's `cancel()`/`uncancel()` counter protocol — no new runtime, no new cancellation model.

2. **Different design philosophy.** anyio's approach is a broad `CancelScope` over the whole operation, with `CancelScope(shield=True)` around the parts that must survive. aiofence takes the inverse: most code runs unaware of cancellation, and you wrap only the expensive, safely-interruptible parts with a `Fence`.

## How It Works

`Fence` is a sync context manager that arms triggers against the current asyncio task. When a trigger fires, the task is cancelled via asyncio's native `cancel()`/`uncancel()` counter protocol. On exit, the `CancelledError` is suppressed and the counter is balanced. The caller inspects `fence.cancelled` and `fence.reasons` after the block. Just asyncio's own machinery, used correctly.

## Requirements

Python 3.12+. No dependencies.

## License

MIT
