Skip to content

Chaining Futures and Transforming Results

Most of the time, transforming an awaitable's result is just value = f(await thing) inside a coroutine. Chaining futures directly — creating a future whose result is derived from another's, wired with done-callbacks rather than a task — is for the places where a coroutine is not available or not wanted: a callback-based library that hands you a future and expects one back, a cache that stores futures so concurrent callers share one in-flight computation, or a bridge between concurrent.futures and asyncio. Getting it right means handling all three outcomes of the source and propagating cancellation in both directions. A helper built this way returned 40 from a source resolved with 4 and a ×10 transform, and cancelling the derived future cancelled its source. asyncio.wrap_future, by contrast, could not stop a thread-pool job that had already started: after cancelling the wrapper, the underlying job was still running.

Prerequisites

1. Check whether a coroutine is simpler

Before chaining by hand, consider the coroutine version. It handles results, exceptions and cancellation automatically:

async def mapped(src: asyncio.Future, fn):
    return fn(await src)


derived = asyncio.ensure_future(mapped(src, parse))

Cancelling derived cancels the task, whose await src then cancels src — the same two-way propagation the hand-written version needs to build. The coroutine costs one task object (about 4–5 µs to create and run on Python 3.14) and one extra loop iteration of latency. Chain by hand when you are inside a synchronous callback that cannot create tasks freely, when you need the derived future to be resolved in the same iteration as the source, or when you are writing a library primitive that should not spawn tasks on its users' behalf.

Verify: if the transform can be expressed as a coroutine and the call site can await, stop here.

2. Build a chain helper that handles every outcome

The source can end three ways, and the derived future must mirror each one. The transform itself can also raise:

import asyncio
from collections.abc import Callable
from typing import TypeVar

T = TypeVar("T")
R = TypeVar("R")


def chain(src: asyncio.Future, fn: Callable[[T], R]) -> asyncio.Future:
    loop = src.get_loop()
    dst: asyncio.Future = loop.create_future()

    def on_src(f: asyncio.Future) -> None:
        if dst.done():                          # dst was cancelled first; nothing to do
            return
        if f.cancelled():
            dst.cancel()
            return
        exc = f.exception()
        if exc is not None:
            dst.set_exception(exc)
            return
        try:
            dst.set_result(fn(f.result()))
        except Exception as e:                  # a failing transform fails dst, not the loop
            dst.set_exception(e)

    def on_dst(f: asyncio.Future) -> None:
        if f.cancelled() and not src.done():
            src.cancel()                        # nobody wants the result any more

    src.add_done_callback(on_src)
    dst.add_done_callback(on_dst)
    return dst

The try around fn is important: without it, an exception from the transform happens inside a done-callback, where it is only logged, and dst is never resolved — every awaiter hangs forever. The dst.done() guard covers the race where dst was cancelled before src completed.

Verify: test four cases — source succeeds, source fails, source cancelled, transform raises — and confirm dst ends in the matching state each time.

Two callbacks wire a derived future to its source A flow of 4 stages. Two callbacks wire a derived future to its source src done result, error or cancel on_src mirror onto dst, run fn safely dst cancelled first caller gave up on_dst cancel src too Forward the outcome downstream and the cancellation upstream; missing either direction leaves work or waiters behind.

3. Propagate cancellation upstream, deliberately

The on_dst callback is what makes the chain behave like await: if everyone waiting on the derived value gives up, the source is cancelled too, so the work stops. Verified: cancelling dst before src completed left src.cancelled() true after one loop iteration.

That is not always right. When the source is shared — a cached in-flight computation that several derived futures hang off — one consumer cancelling must not cancel the source for the others:

class SharedFetch:
    def __init__(self) -> None:
        self._inflight: dict[str, asyncio.Future] = {}

    def get(self, key: str, parse) -> asyncio.Future:
        src = self._inflight.get(key)
        if src is None:
            src = asyncio.ensure_future(fetch(key))
            self._inflight[key] = src
            src.add_done_callback(lambda _: self._inflight.pop(key, None))
        return chain_no_upstream_cancel(src, parse)     # consumers cannot cancel the shared src

chain_no_upstream_cancel is chain without the on_dst callback. Shared sources then need their own lifetime policy — cancel when the last consumer leaves, or let them finish and populate a cache. This is the same trade-off as shielding in implementing the single-flight pattern.

Verify: with two consumers on one shared source, cancelling one leaves the other's result intact.

Should cancelling the derived future cancel the source? A grid of 3 rows by 3 columns. Should cancelling the derived future cancel the source? source is cancel upstream? mechanism private to one consumer yes on_dst cancels src shared by many consumers no chain without on_dst shared, expensive when the last one leaves reference count The direction of cancellation is a policy decision, not a detail.

4. Know what wrap_future can and cannot stop

asyncio.wrap_future(cf_future) chains a concurrent.futures.Future to an asyncio future, with thread-safe callbacks in both directions. Cancelling the asyncio side calls cancel() on the concurrent future — which only succeeds if the job has not started yet:

import concurrent.futures as cf
import time


async def demo() -> None:
    with cf.ThreadPoolExecutor() as pool:
        job = pool.submit(time.sleep, 0.2)
        wrapped = asyncio.wrap_future(job)
        wrapped.cancel()
        await asyncio.sleep(0.01)
        print(job.cancelled(), job.running())     # False True

Verified on Python 3.14: after cancelling the wrapper, the thread job reported cancelled() == False and running() == True. The asyncio side is cancelled; the thread keeps going and its result is discarded. This is the same limitation as asyncio.to_thread cancellation, and the fix is the same — cooperative cancellation inside the job, as in cancelling asyncio.to_thread calls.

Verify: for every wrap_future call site, decide what should happen to the thread job when the awaiter is cancelled, and implement it with a stop flag if "keep running" is wrong.

5. Chain several steps without a tower of callbacks

A pipeline of transforms becomes a chain of chains. Keep it flat by folding over the steps, and keep error context by wrapping transform failures:

from functools import reduce


class StepError(Exception):
    def __init__(self, step: str, cause: BaseException) -> None:
        super().__init__(f"{step} failed: {cause!r}")
        self.step = step


def named(step: str, fn):
    def wrapper(value):
        try:
            return fn(value)
        except Exception as e:
            raise StepError(step, e) from e
    return wrapper


def pipeline(src: asyncio.Future, *steps) -> asyncio.Future:
    return reduce(lambda fut, s: chain(fut, named(s.__name__, s)), steps, src)


result = await pipeline(raw_bytes_future, decode, parse_json, validate)

Each link propagates cancellation upstream through the whole chain, so cancelling the final future cancels the source. Each failure carries the name of the step that failed. Past three or four steps, though, an async def that awaits once and calls the functions in sequence is shorter, equally fast, and easier to debug — chains are a tool for the boundary between callback code and coroutines, not a replacement for them.

Verify: make the middle step raise; the final future fails with StepError naming that step, and the source is not left pending.

Chain futures, or write a coroutine? A decision on Where does the transform happen with 3 outcomes. Chain futures, or write a coroutine? Where does the transform happen? in async code a coroutine await, then transform in sync callback code chain() done-callbacks only on a thread pool future wrap_future + chain job may keep running Hand-wired chains belong at boundaries; inside async code, await is clearer.

Verification

A future chain is correct when:

  • All four outcomes — success, failure, cancellation, transform error — resolve the derived future.
  • No awaiter hangs because a transform raised inside a callback.
  • Upstream cancellation matches the sharing policy: private sources cancelled, shared ones protected.
  • Thread-backed sources stop, or are known to keep running, when awaiters give up.

Diagnostic Hook: count derived futures by final state, and alert on any that remain pending after their source is done — that means a callback path forgot to resolve them. For wrap_future call sites, track the number of thread jobs still running after their awaiter was cancelled; a growing number is thread-pool capacity spent on work nobody wants.

Pitfalls & edge cases

  • Raising from a transform inside a callback without catching it: the derived future never resolves.
  • Calling set_result on an already-cancelled derived future: InvalidStateError; check done() first.
  • Creating the derived future on a different loop from the source: callbacks then cross loops unsafely; use src.get_loop().
  • Assuming wrap_future cancellation stops the thread: it only prevents jobs that have not started.

Frequently Asked Questions

How do I transform the result of an asyncio Future?

In async code, await it and transform the value, or wrap that in a small coroutine and ensure_future it. Without a coroutine, create a new future and add a done-callback to the source that sets the transformed result, forwarding exceptions and cancellation.

Does cancelling a chained future cancel the original?

Only if you wire it: add a done-callback on the derived future that cancels the source when the derived one is cancelled. Do this for private sources, not for sources shared by several consumers.

Does cancelling asyncio.wrap_future stop the thread?

No, not once the job has started. It cancels the asyncio future and tries to cancel the concurrent future, which only succeeds for jobs still waiting in the pool queue.

Why does my chained future never resolve?

Usually the transform raised inside a done-callback. That exception is only logged by the loop, and the derived future is never set. Catch exceptions from the transform and set them on the derived future.