ryou 0.1.0

Go-style concurrency for Python. No async/await. No brokers. No boilerplate.

ryou

Go-style concurrency for Python. No async/await. No brokers. No boilerplate.

from ryou import ry, Chan, wait

@ry
def fetch(url: str) -> str:
    ...

h = ~fetch("https://example.com")
result, err = h.result()
if err:
    raise err
print(result)

Install

pip install ryou

Requires Python 3.12+, Linux/macOS only.

Concepts

Three primitives:

• @ry — turns any function into a ryoutine. ~fn() spawns it concurrently.
• Ry[T] — handle to a running ryoutine. .result() blocks until done, always returns (value, err).
• Chan[T] — typed channel. send/recv block naturally, no busy loops.

Powered by gevent greenlets under the hood. IO ryoutines are cooperative greenlets. CPU ryoutines are forked processes. You never interact with gevent directly.

Quick Start

from ryou import ry, wait

@ry
def fetch(url: str) -> str:
    import urllib.request
    return urllib.request.urlopen(url).read(100).decode()

# spawn concurrently
h1 = ~fetch("https://example.com")
h2 = ~fetch("https://httpbin.org/html")

wait(h1, h2)

result, err = h1.result()
if err:
    print("failed:", err)
else:
    print(result)

~fn() spawns and returns a Ry[T] handle immediately. .result() blocks until the ryoutine finishes.

Channels

from ryou import ry, Chan, wait

ch = Chan[str]()

@ry
def producer():
    for i in range(5):
        ch.send(f"message {i}")
    ch.send(None)  # signal done

@ry
def consumer():
    while (msg := ch.recv()) is not None:
        print("got:", msg)

wait(~producer(), ~consumer())

Chan[T] is a typed, blocking queue. send blocks if full, recv blocks until an item is available. Goroutines yield while waiting — no busy loops.

Bounded channel:

ch = Chan[str](maxsize=10)  # blocks sender when full

CPU Bound Work

from ryou import ry, wait

@ry(cpu=True)
def crunch(n: int) -> int:
    return sum(range(n))

h = ~crunch(10**8)
result, err = h.result()
print(result)

cpu=True runs the function in a forked child process — true parallelism, no GIL. IO and CPU ryoutines can run simultaneously.

Error Handling

.result() never raises. It always returns (value, err):

h = ~fetch("https://bad-url")
result, err = h.result()
if err:
    print("error:", err)  # handle it
else:
    print(result)

Wait

wait()           # wait for all running ryoutines
wait(h1, h2)     # wait for specific handles

API Reference

@ry

@ry
def fn() -> T: ...          # IO bound, runs on greenlet

@ry(cpu=True)
def fn() -> T: ...          # CPU bound, runs in forked process

Ry[T]

h = ~fn()                              # spawn
val, err = h.result()                  # block until done, (T | None, Exception | None)

Chan[T]

ch = Chan[T]()              # unbounded
ch = Chan[T](maxsize=10)    # bounded

ch.send(value)              # blocks if full
value = ch.recv()           # blocks until item available

wait

wait()          # wait for all
wait(h1, h2)    # wait for specific handles

License

MIT