rustypie 0.1.0

Rust-inspired implementations of Result<T, E>, Option<T> and Iter<T> data types for Python.

rusty

Rust-inspired Result<T, E>, Option<T>, and Iter<T> types for Python.

rusty is a single, dependency-free module that brings Rust's most useful data-modeling patterns to Python: explicit error contracts alongside Python's exception mechanism, optional values instead of ambiguous None checks, and lazy, chainable iterators instead of nested comprehensions.

Why rusty?

• Explicit error contracts. A Result-returning API makes modeled failure paths visible in its signature. Callers explicitly choose whether to propagate, transform, recover from, or unwrap the result.
• Composable by design. map, and_then, or_else, filter, and friends let you build pipelines without intermediate if/else chains.
• Familiar to Rust developers. Method names and core workflows closely follow std::result::Result, std::option::Option, and Iterator, with Python-friendly adaptations.
• Zero dependencies, one file. Drop rusty.py into any project running Python 3.12+.
• Fully tested. The test suite covers all library features.

Installation

rusty has no runtime dependencies. Its PyPI distribution is named rustypie and can be installed with pip:

pip install rustypie

The installed module is imported as rusty.

Alternatively, since the entire library is a single file, you can simply copy rusty.py into your project.

Table of Contents

• Result: Ok / Err
• AsyncResult
• Option
• Iter
• Collecting Result Values
• Decorators
• Quick Reference
• Testing
• License

Result: Ok / Err

Result[T, E] is a Union[Ok[T], Err[E]] representing either a success value or an error value as a visible return-value contract.

from rusty import Err, Ok, Result


def divide(a: int, b: int) -> Result[float, str]:
    if b == 0:
        return Err("division by zero")
    return Ok(a / b)


result = divide(10, 2)
assert result.is_ok()
assert result.unwrap() == 5.0

doubled = divide(10, 2).map(lambda value: value * 2).unwrap_or(0.0)
assert doubled == 10.0

# Errors short-circuit the chain instead of raising:
failure = divide(10, 0).map(lambda value: value * 2).unwrap_or(-1.0)
assert failure == -1.0

Ok and Err support Python's structural pattern matching too:

match divide(10, 0):
    case Ok(value):
        print(f"got {value}")
    case Err(message):
        print(f"failed: {message}")

Key methods available on both Ok and Err:

Method	Description
is_ok() / is_err()	Check the variant.
unwrap()	Extract the success value, or raise on Err.
expect(msg)	Extract the success value, or raise RuntimeError(msg) on Err.
unwrap_or(default) / unwrap_or_else(fn)	Extract the value or fall back.
map(fn) / map_err(fn)	Transform the success or error value.
map_or(default, fn) / map_or_else(default_fn, fn)	Transform with a fallback.
and_then(fn) / or_else(fn)	Chain fallible operations.
op_and(other) / op_or(other)	Rust's and/or combinators.
ok() / err()	Convert to Option[T] / Option[E].
as_ok() / as_err()	Get self or None without unwrapping.
inspect(fn) / inspect_err(fn)	Peek at the value for side effects.
iter()	Get an Iter yielding zero or one item.

unwrap() returns the value from Ok. On Err, it re-raises the wrapped value when that value is an Exception; otherwise it raises a RuntimeError describing the error. expect(msg) also returns the value from Ok, but always raises RuntimeError(msg) on Err. These methods are intentional escape hatches for callers that explicitly choose exception-based handling at a boundary. Exceptions raised independently by user callbacks still propagate normally.

AsyncResult

AsyncResult[T, E] wraps an Awaitable[Result[T, E]], letting you chain map, and_then, map_err, and or_else — synchronously or with async callbacks — before finally await-ing the resolved Result. Once the awaitable resolves to a Result, that value is cached and returned by later awaits.

from rusty import Ok, Result, async_result


async def fetch_value() -> Result[int, str]:
    return Ok(21)


async def transformed_value() -> Result[int, str]:
    return await (
        async_result(fetch_value())
        .map(lambda value: value * 2)
        .and_then(lambda value: Ok(value + 1))
    )

map, map_err, and_then, and or_else accept plain callables or ones returning awaitables, so sync and async steps can be mixed freely in the same chain. AsyncResult also exposes async ok(), err(), as_ok(), as_err(), unwrap_or_else(), and iter() for resolving into Option/Iter values.

Option

Option[T] represents an optional value (Some(value) or None) without Python's ambiguity around whether a bare None means "absent" or "an actual None value".

from rusty import Option


name = Option.some("Ferris").map(str.upper).unwrap_or("UNKNOWN")
assert name == "FERRIS"

missing = Option.none().map(str.upper).unwrap_or("UNKNOWN")
assert missing == "UNKNOWN"

Notable methods:

Method	Description
is_some() / is_none()	Check the variant.
unwrap()	Extract the value, or raise ValueError on None.
expect(msg)	Extract the value, or raise ValueError(msg) on None.
unwrap_or(default) / unwrap_or_else(fn)	Extract with a fallback.
map(fn), map_or(default, fn), map_or_else(default_fn, fn)	Transform the value.
and_then(fn) / or_else(fn)	Chain optional-returning operations.
op_and(other) / op_or(other) / op_xor(other)	Combine two Option values.
filter(predicate)	Keep the value only if it matches a predicate.
zip(other) / zip_with(other, fn)	Combine two Option values.
ok_or(err)	Convert to Result[T, E].
take()	Move the value out, leaving None behind.
iter()	Get an Iter yielding zero or one item.

Option.some(value) raises ValueError if value is None — use Option.none() (or the Option(value) constructor) to represent absence explicitly. For an absent value, unwrap() raises a standard descriptive ValueError, while expect(msg) raises ValueError with the supplied message.

Iter

Iter[T] wraps any Iterable[T] with lazy, chainable combinators inspired by Rust's Iterator trait. Nothing is evaluated until a terminal method (like collect(), fold(), or for_each()) consumes the iterator.

from rusty import into_iter


values = (
    into_iter([1, 2, 3, 4, 5, 6])
    .filter(lambda value: value % 2 == 0)
    .map(lambda value: value * 10)
    .collect()
)
assert values == [20, 40, 60]

Transformations return a new lazy Iter:

Method	Description
map(fn) / filter(predicate) / filter_map(fn)	Transform or prune elements.
flat_map(fn) / flatten()	Flatten one level of Iterable, Result, or Option.
take(n) / take_while(predicate)	Limit the front of the iterator.
skip(n) / skip_while(predicate)	Drop from the front of the iterator.
chain(other) / zip(other)	Combine with another iterable.
enumerate()	Pair each item with its index.
windows(size)	Yield overlapping tuples of size (like slice::windows).
inspect(fn)	Run a side effect as each item is consumed, yielding it unchanged.

Terminal methods consume the iterator and return a concrete result:

Method	Description
collect() / collect_set() / collect_tuple()	Materialize into a list, set, or tuple.
fold(init, fn) / reduce(fn)	Accumulate into a single value.
all(predicate) / any(predicate)	Boolean tests over all elements.
find(predicate) / position(predicate) / rposition(predicate)	Locate a matching element.
count() / last() / nth(n) / next()	Positional access.
partition(predicate)	Split into (matched, unmatched) lists.
for_each(fn)	Run a side effect per item, returning None.

flatten() and flat_map() understand Ok, Err, and Option in addition to plain iterables, so you can flatten a stream of Result/Option values directly:

from rusty import Err, Iter, Ok


assert Iter([Ok(1), Err("skip me"), Ok(3)]).flatten().collect() == [1, 3]

Collecting Result Values

collect_ok, collect_ok_set, and collect_ok_tuple turn an Iter[Result[T, E]] (or any iterable of Result-like objects) into a single Result. They short-circuit and return the first Err encountered, or an Ok containing all successfully unwrapped values:

from rusty import Err, Iter, Ok, collect_ok, collect_ok_set, collect_ok_tuple


values = [Ok(1), Ok(2), Ok(3)]

assert collect_ok(Iter(values)) == Ok([1, 2, 3])
assert collect_ok_set(Iter(values)) == Ok({1, 2, 3})
assert collect_ok_tuple(Iter(values)) == Ok((1, 2, 3))

assert collect_ok(Iter([Ok(1), Err("invalid"), Ok(3)])) == Err("invalid")

Iteration stops as soon as an Err is found, so any remaining items are never evaluated.

Decorators

Wrap existing exception-raising or None-returning functions (or methods) without rewriting their bodies:

Decorator	Use for	Wraps into
@resultify	Functions that may raise an exception.	Result[T, Exception]
@resultify_method	Instance methods that may raise an exception.	Result[T, Exception]
@optionable	Functions that may return None.	Option[T]
@optionable_method	Instance methods that may return None.	Option[T]

from typing import Optional

from rusty import optionable, resultify


@resultify
def parse_int(value: str) -> int:
    return int(value)


assert parse_int("42").map(lambda n: n * 2).unwrap() == 84
assert parse_int("nope").is_err()


@optionable
def find_user(user_id: int) -> Optional[str]:
    return {1: "Ferris"}.get(user_id)


assert find_user(1).unwrap_or("unknown") == "Ferris"
assert find_user(2).unwrap_or("unknown") == "unknown"

Quick Reference

from rusty import (
    AsyncResult,
    Err,
    Iter,
    Ok,
    Option,
    Result,
    async_result,
    collect_ok,
    collect_ok_set,
    collect_ok_tuple,
    into_iter,
    optionable,
    optionable_method,
    resultify,
    resultify_method,
)

Testing

The project uses pytest. Install the development dependencies and run the test suite from the repository root:

python -m pip install -e ".[dev]"
python -m pytest

License

MIT License. See LICENSE.