ryou 0.2.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

Five 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.
• select — wait on multiple channels, pick whichever fires first.
• Context — cancellation and timeout propagation.

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.close()  # signal done

@ry
def consumer():
    for msg in ch:  # exits when channel is closed
        print("got:", msg)

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

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

Bounded channel:

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

Sending on a closed channel raises RuntimeError. Receiving from a closed empty channel returns CLOSED.

Select

Wait on multiple channels — pick whichever fires first:

from ryou import ry, Chan, select, wait, CLOSED

ch1 = Chan[str]()
ch2 = Chan[str]()

@ry
def handler():
    active = {ch1, ch2}
    while active:
        ch, msg = select(*active)
        match (ch, msg):
            case (_, msg) if isinstance(msg, type(CLOSED)):
                active.remove(ch)
            case (c, v) if c is ch1:
                print(f"ch1: {v}")
            case (c, v) if c is ch2:
                print(f"ch2: {v}")

select is non-deterministic when multiple channels are ready simultaneously — same as Go.

Context

Cancel a ryoutine from outside:

from ryou import ry, Context, wait

@ry
def worker(ctx=None):
    for i in range(10):
        if ctx and ctx.cancelled():
            return
        time.sleep(0.5)
        print(f"tick {i}")

ctx = Context()
h = ~worker(ctx=ctx)

@ry
def canceller():
    time.sleep(1.5)
    ctx.cancel()

wait(~canceller(), h)
print(ctx.err())  # "cancelled"

Auto-cancel with timeout:

ctx = Context(timeout=1.5)
h = ~worker(ctx=ctx)
wait(h)
print(ctx.err())  # "timeout"

ctx.cancel() kills the ryoutine immediately. ctx.cancelled() lets the ryoutine check cooperatively.

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)
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, raises RuntimeError if closed
value = ch.recv()            # blocks until item available, returns CLOSED if closed + empty
ch.close()                   # close channel, wakes blocked receivers
ch.done()                    # True if closed

for msg in ch: ...           # drain until closed, must be inside @ry

select

ch, msg = select(ch1, ch2)   # blocks until one fires, returns (channel, value)

Context

ctx = Context()              # no timeout
ctx = Context(timeout=5.0)   # auto cancel after 5s

ctx.cancel()                 # forceful kill
ctx.cancelled()              # cooperative check
ctx.err()                    # "cancelled" | "timeout" | None

wait

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

License

MIT