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
• Unit: successes without a value
• 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:

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.

Unit: successes without a value

Python's None carries two meanings: "no meaningful value" and "absent". Option uses it for the latter — absence is _value is None — so a None payload cannot survive the trip into an Option:

from rusty import UNIT, Ok, Option

assert Ok(None).ok() == Option.none()          # collapses: None means absent
assert Ok(UNIT).ok() == Option.some(UNIT)      # survives: the success is visible

UNIT is the singleton of Unit, the equivalent of Rust's (). Reach for it when an operation succeeds but has nothing to hand back — Result[Unit, E] reads as "worked, no payload", and stays distinguishable after .ok():

from rusty import UNIT, Err, Ok, Result, Unit


def save(record: dict) -> Result[Unit, str]:
    if not record:
        return Err("empty record")
    ...  # perform the write
    return Ok(UNIT)


assert save({"id": 1}).ok().is_some()

Unit is a slotted singleton: Unit() always returns UNIT, identity holds across pickling and copying, and it adds no per-instance dictionary.

Note that Ok(None) remains perfectly valid and no longer raises anywhere — converting it to an Option simply yields Option.none(). The same applies to any callback that returns None, such as Option.map(fn) or Iter.reduce(fn).

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:

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:

Terminal methods consume the iterator and return a concrete result:

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:

@resultify and @resultify_method accept none_as_unit (default False). When enabled, a None return becomes Ok(UNIT) instead of Ok(None), so the success stays visible after .ok(). It is off by default because a None return is often meaningful data — dict.get and re.match return None as a value, and only the caller can tell the two cases apart:

from rusty import UNIT, Ok, resultify


@resultify
def clear_cache() -> None: ...


@resultify(none_as_unit=True)
def clear_cache_unit() -> None: ...


assert clear_cache() == Ok(None)          # unchanged default behaviour
assert clear_cache_unit() == Ok(UNIT)     # opt in where None means "nothing"

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 (
    UNIT,
    AsyncResult,
    Err,
    Iter,
    Ok,
    Option,
    Result,
    Unit,
    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.