python-coroutine 0.1.0

A lightweight, zero-dependency concurrency library with Future[T]

coro

A lightweight, zero-dependency coroutine library for Python. Wraps ThreadPoolExecutor behind an ergonomic Future[T] API that works with both sync and async functions.

import httpx
import coro

@coro.future
async def fetch(url: str) -> int:
    async with httpx.AsyncClient() as client:
        response = await client.get(url)
        return response.status_code

@coro.main
def main():
    status = fetch("https://httpbin.org/get").map(str).result()
    print(status)  # "200"

Install

With astral-uv

uv add coro

Or with pip

pip install coro

API

coro.run(fn)

Runs a function in a new runtime and returns its result. Blocks until complete.

result = coro.run(lambda: "hello")

Works with both sync and async functions:

result = coro.run(some_async_function)

coro.spawn(fn, *args, **kwargs)

Schedules a function on the default runtime and returns a Future[T] immediately. Works with both sync and async defs.

fut = coro.spawn(fetch, "https://example.com")
# do other work...
result = fut.result()

coro.future

Decorator that transforms any function (sync or async) into one that returns a Future[T].

@coro.future
def compute(n: int) -> int:
    return n * n

fut = compute(42)

coro.main

Decorator that replaces the if __name__ == "__main__": boilerplate. When the module is run directly, it calls coro.run(fn) automatically. When imported, the function is returned as-is.

@coro.main
def entry():
    print("hello")

Future[T]

The core primitive. Created by spawn, join_all, select, sleep, and the @coro.future decorator.

Method	Description
.result()	Block until resolved, return the value (or raise)
.done()	Non-blocking check if the future has resolved
.map(fn)	Transform the result via fn — returns Future[U]
.and_then(fn)	Chain with fn: T -> Future[U] — returns Future[U]
.timeout(seconds)	Raise TimeoutError if not resolved in time
.add_done_callback(cb)	Call cb(future) when resolved

coro.join_all(*futures)

Wait for all futures to complete, return Future[list[...]].

f1 = spawn(fetch, "https://a.com")
f2 = spawn(fetch, "https://b.com")
results = join_all(f1, f2).result()

coro.select(*futures)

Wait for the first future to complete, return Future[tuple[int, T]] with the index and value.

idx, value = select(fast, medium, slow).result()

coro.sleep(seconds)

Non-blocking sleep that returns a Future[None].

sleep(1.5).result()

coro.Runtime

Manages the thread pool. Created automatically via the module-level functions, or you can create your own.

rt = Runtime(max_workers=8)
fut = rt.spawn(fetch, "https://example.com")
result = fut.result()
rt.shutdown()

How it works

spawn submits work to a ThreadPoolExecutor. If the function is an async def, it's wrapped with asyncio.run(). The Future[T] uses threading.Event for blocking the caller and threading.Lock for thread-safe completion gating.

Requirements

Python 3.14+ (uses PEP 695 generic syntax, @overload with type params, and Callable[P, T]).

License

MIT