Async Helpers

Async Helpers (sillo.core.helpers.async_helpers)

from sillo.core.helpers.async_helpers import is_async_callable
from sillo.core.helpers.async_helpers import AwaitableOrContextManagerWrapper
from sillo.core.helpers.async_helpers import collapse_excgroups

Import from sillo.core.helpers, not sillo.helpers

This module lives under sillo.core.helpers. The shorter sillo.helpers.async_helpers path does not exist and raises ModuleNotFoundError. The same applies to Deprecation; every other module in this section is under sillo.helpers.

These helpers are the smallest, most internal utilities in sillo. They solve two problems that appear everywhere in an ASGI framework:

1. Detecting whether something is an async callable — not just a plain function, but also objects whose __call__ is a coroutine function, and functools.partial wrappers around either.
2. Adapting a coroutine that produces a closeable resource into something you can both await and use as an async with block, so the resource is always closed when the block exits.

Nothing here depends on third-party packages. It is pure standard library (asyncio, functools, contextlib) plus a guarded import of exceptiongroup on Python < 3.11.

is_async_callable

The most-used helper. It returns True when obj can be awaited as a call — either because it is a coroutine function, or because calling it produces a coroutine.

from sillo.core.helpers.async_helpers import is_async_callable


async def handler():
    return "ok"


def sync_handler():
    return "ok"


print(is_async_callable(handler))       # True
print(is_async_callable(sync_handler))  # False

Why this is not just asyncio.iscoroutinefunction

Two cases break a naive check:

• Objects with an async __call__. A class instance can be “callable and async” even though asyncio.iscoroutinefunction(instance) is False. sillo checks obj.__call__ as a fallback.

  from sillo.core.helpers.async_helpers import is_async_callable
  
  
  class Controller:
      async def __call__(self):
          return "dispatched"
  
  
  controller = Controller()
  print(is_async_callable(controller))  # True

• functools.partial. Wrapping an async function in partial (common for binding default arguments) hides the underlying coroutine function. sillo unwraps partial layers before testing.

  from functools import partial
  from sillo.core.helpers.async_helpers import is_async_callable
  
  
  async def greet(name):
      return f"hi {name}"
  
  
  bound = partial(greet, "world")
  print(is_async_callable(bound))  # True

Full signature and typing

def is_async_callable(obj: typing.Any) -> TypeGuard[AwaitableCallable[typing.Any]]: ...

It is declared with @typing.overload so type checkers narrow correctly: passing a known AwaitableCallable keeps the AwaitableCallable type; passing Any narrows to the awaitable-callable type guard. Internally the logic is:

def is_async_callable(obj):
    while isinstance(obj, functools.partial):
        obj = obj.func
    return asyncio.iscoroutinefunction(obj) or (
        callable(obj) and asyncio.iscoroutinefunction(obj.__call__)
    )

Where sillo uses it

• sillo.application — deciding whether a user-supplied handler runs in the async event loop or can be wrapped.
• sillo.routing.router — distinguishing async route endpoints from sync ones.
• sillo.http.request — checking whether request-dependent callables are awaitable before scheduling them.
• sillo.testclient._internal.utils — the test client reuses the exact same helper so test behavior matches production.

A practical pattern

When you accept a callback from a user and must decide how to invoke it:

from sillo.core.helpers.async_helpers import is_async_callable


async def run_maybe_async(fn, *args):
    if is_async_callable(fn):
        return await fn(*args)
    return fn(*args)

AwaitableOrContextManagerWrapper

Some coroutines return a resource that must be closed (a connection, a session, a stream). You want to write code that either:

resource = await open_thing()      # await it directly
await resource.close()

or:

async with open_thing() as resource:   # use it as a context manager
    ...

AwaitableOrContextManagerWrapper lets one object support both forms. __await__ returns the underlying resource; __aenter__/__aexit__ enter the context and call .close() on exit.

from sillo.core.helpers.async_helpers import AwaitableOrContextManagerWrapper


class Connection:
    async def close(self):
        print("connection closed")


async def open_connection():
    return Connection()


# Wrap the coroutine so it is both awaitable and an async context manager
awaiter = AwaitableOrContextManagerWrapper(open_connection())

# Form 1: await directly
conn = await awaiter
print(type(conn).__name__)  # Connection

# Form 2: async with (re-wrap a fresh coroutine)
async with AwaitableOrContextManagerWrapper(open_connection()) as conn2:
    print(type(conn2).__name__)  # Connection
# "connection closed" printed automatically on exit

The supporting protocols

Two typing.Protocol types describe the contract so type checkers understand the dual nature:

class AwaitableOrContextManager(Awaitable[T_co], AsyncContextManager[T_co], Protocol[T_co]): ...

This is the static type of “something you can await and use as an async context manager.” AwaitableOrContextManagerWrapper is the runtime adapter that turns a plain Awaitable[Closeable] into that shape.

There is also:

class SupportsAsyncClose(Protocol):
    async def close(self) -> None: ...

Used as the bound for the wrapper’s generic parameter, so the wrapper only accepts awaitables that resolve to something with an async close().

Implementation notes

• Uses __slots__ = ("aw", "entered") to keep wrapper instances light.
• __aexit__ always calls self.entered.close() and returns None, meaning it does not suppress exceptions — any error raised inside the async with block propagates after the resource is closed.
• If you await the wrapper and then also want the context-manager form, you must wrap a fresh coroutine; the same wrapper instance tracks its entered resource once.

Where sillo uses it

AwaitableOrContextManager and its wrapper appear in request/response handling where a body stream may be consumed either by awaiting it or by entering it as a context manager. sillo.http.request imports AwaitableOrContextManager and AwaitableOrContextManagerWrapper directly.

collapse_excgroups

On Python 3.11+, BaseExceptionGroup lets multiple exceptions be raised together. In some paths sillo would rather surface the single underlying exception than a group that wraps exactly one item. collapse_excgroups is a context manager that unwraps single-element exception groups.

from sillo.core.helpers.async_helpers import collapse_excgroups


try:
    with collapse_excgroups():
        raise BaseExceptionGroup("one", [ValueError("boom")])
except ValueError as exc:
    print(exc)  # ValueError: boom  (not the group)

How it works

@contextmanager
def collapse_excgroups():
    try:
        yield
    except BaseException as exc:
        if has_exceptiongroups:
            while isinstance(exc, BaseExceptionGroup) and len(exc.exceptions) == 1:
                exc = exc.exceptions[0]
        raise exc

has_exceptiongroups is True on Python 3.11+, and on older versions it is True only if the exceptiongroup backport is installed (otherwise False, and the group is re-raised untouched). The unwrap loop continues while the group contains exactly one exception, so a group-of-one-of-one collapses fully.

Where sillo uses it

sillo._internals._middleware wraps middleware execution in collapse_excgroups so that a middleware raising a single exception is reported as that exception, not as a noisy BaseExceptionGroup.

Why sync and async cannot be treated the same

The reason this module exists is that Python gives you no way to call a function without first knowing which kind it is.

Calling an async function does not run it. It builds a coroutine object and returns immediately:

async def work():
    print("running")

work()          # prints nothing, returns <coroutine object>
await work()    # prints "running"

Awaiting a sync function fails the other way:

def work():
    return "done"

await work()    # TypeError: object str can't be used in 'await' expression

So dispatching code that accepts arbitrary callables must branch, and the branch has to be right for every shape a callable can take. That is three shapes, not one: a plain async def, an instance whose __call__ is async def, and a functools.partial wrapping either. is_async_callable collapses all three into one boolean.

The consequence of getting it wrong is worth stating plainly. If you call an async function without awaiting, nothing runs and nothing raises — you get a RuntimeWarning on stderr and silently missing behaviour. That is a bug that reaches production, because the code path “works” in the sense that it returns without error.

When you actually need these

Most application code never imports this module. sillo calls is_async_callable for you when it dispatches a handler, and the framework wires up the wrapper and the exception-group collapse internally. You reach for these directly in three situations:

You are writing a plugin or an extension point. Any API that accepts a user-supplied callback has to decide whether to await the result. Hardcoding one or the other forces your users into a style they may not want.

You are building middleware that wraps arbitrary handlers. Middleware sits between the framework and code it does not control, so it inherits the same dispatch problem.

You are writing a library that returns a closeable resource. The wrapper lets callers choose between await and async with without you shipping two APIs.

If you are writing route handlers and business logic, you will not need any of it.

Anti-patterns

Do not call an async function without awaiting it

The single most common async bug in Python, and it fails silently:

async def send_email(to: str): ...

# Wrong: creates a coroutine object and throws it away.
send_email("a@example.com")
# RuntimeWarning: coroutine 'send_email' was never awaited

Nothing happens. No email, no exception, just a warning on stderr that is easy to miss in production logs. The is_async_callable check exists so dispatching code never makes this mistake:

result = await fn(*args) if is_async_callable(fn) else fn(*args)

Do not use callable() or inspect.iscoroutinefunction on their own

callable(obj) is true for both sync and async, so it tells you nothing about dispatch.

asyncio.iscoroutinefunction(obj) is closer but returns False for two shapes sillo supports: a class instance whose __call__ is async, and a functools.partial wrapping an async function. Both appear constantly in framework code — class-based views are the first, and binding default arguments to a handler is the second.

is_async_callable handles all three. Use it rather than reimplementing the check.

Do not reuse an AwaitableOrContextManagerWrapper instance

A wrapper holds one coroutine and tracks one entered resource. Awaiting it and then entering it as a context manager reuses a coroutine that has already run, which raises RuntimeError: cannot reuse already awaited coroutine.

Wrap a fresh coroutine each time:

# Wrong
awaiter = AwaitableOrContextManagerWrapper(open_connection())
conn = await awaiter
async with awaiter as conn2:      # RuntimeError
    ...

# Right
conn = await AwaitableOrContextManagerWrapper(open_connection())
async with AwaitableOrContextManagerWrapper(open_connection()) as conn2:
    ...

__aexit__ does not suppress exceptions

The wrapper closes the resource and then lets the exception propagate. That is the correct default — silently swallowing errors inside an async with block hides bugs — but it means you still need your own try/except if a failure is expected.

Cleanup is guaranteed either way: the resource is closed whether the block exits normally or by exception.

collapse_excgroups only collapses groups of exactly one

A group containing two or more exceptions is re-raised untouched, because there is no single exception to collapse to. Code that catches a specific type after this context manager still needs to handle BaseExceptionGroup for the multi-error case:

try:
    with collapse_excgroups():
        await run_concurrent_tasks()
except ValueError:
    ...                      # one task failed
except BaseExceptionGroup:
    ...                      # several failed at once

On Python below 3.11 without the exceptiongroup backport installed, has_exceptiongroups is False and nothing is collapsed at all. Do not rely on the collapse for correctness on older runtimes.

Performance

is_async_callable is a while loop over partial layers plus one or two iscoroutinefunction checks. That is nanoseconds, but it is not free, and it runs on a value that never changes for a given handler.

sillo resolves it once at route registration rather than per request, which is why handler dispatch has no introspection cost at request time. If you write your own dispatcher, do the same:

class Dispatcher:
    def __init__(self, fn):
        self.fn = fn
        self.is_async = is_async_callable(fn)   # once, at construction

    async def __call__(self, *args):
        return await self.fn(*args) if self.is_async else self.fn(*args)

Checking inside the hot path instead means paying for introspection on every request to compute an answer that was decided when the route was registered.

AwaitableOrContextManagerWrapper uses __slots__, so instances are small and allocation is cheap. collapse_excgroups costs nothing on the success path — the try block is free in CPython until an exception is actually raised.

Summary

Helper	Purpose	Returns / Type
is_async_callable(obj)	Detect async callables (incl. partial, async __call__)	bool (type-guarded)
AwaitableOrContextManager[T]	Protocol: awaitable and async context manager	Protocol
SupportsAsyncClose	Protocol: has async def close()	Protocol
AwaitableOrContextManagerWrapper(aw)	Adapter: coroutine → awaitable + async CM	wrapper instance
collapse_excgroups()	Context manager unwrapping single-exception groups	context manager

Related

• Handlers — how sillo dispatches sync and async route handlers
• Middleware — the main place these helpers appear in user code
• Concurrency — running blocking work without stalling the loop
• Retry — uses is_async_callable to pick a wrapper
• Request Lifecycle — where dispatch decisions are made