Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
57 changes: 48 additions & 9 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -56,7 +56,7 @@ You do not need to switch everything at once. A realistic path is:

### 1) Result and Option

Rust-style `Result[T, E]` and `Option[T]` as first-class values.
Rust-style `Result[T]` and `Option[T]` as first-class values.

```python
from pyropust import Ok, Err, Some, None_
Expand All @@ -68,12 +68,12 @@ maybe = Some(42)
empty = None_()
```

Result is explicit about failures. You can return it from functions and branch on `is_ok / is_err` without exceptions.
Result is explicit about failures. All failures are represented as `RopustError`. You can return it from functions and branch on `is_ok / is_err` without exceptions.

```python
from pyropust import Ok, Err, Result

def divide(a: int, b: int) -> Result[float, str]:
def divide(a: int, b: int) -> Result[float]:
if b == 0:
return Err("division by zero")
return Ok(a / b)
Expand All @@ -83,6 +83,7 @@ if res.is_ok():
value = res.unwrap()
else:
error = res.unwrap_err()
print(error.message)
```

Keep Option short and explicit: you must unwrap or provide defaults.
Expand Down Expand Up @@ -111,8 +112,8 @@ from pyropust import Ok

res = (
Ok("123")
.map(int) # Result[int, E]
.map(lambda x: x * 2) # Result[int, E]
.map(int) # Result[int]
.map(lambda x: x * 2) # Result[int]
.and_then(lambda x: Ok(f"Value is {x}"))
)
print(res.unwrap()) # "Value is 246"
Expand All @@ -121,21 +122,58 @@ print(res.unwrap()) # "Value is 246"
When to use: `map/and_then` is best for small, expression-style transforms where each step is a function.

> [!TIP]
> **Type Hint for `and_then`**: When using `and_then` with a callback that may return `Err`, define the initial `Result` with an explicit return type annotation. This ensures the error type is correctly inferred.
> **Type Hint for `and_then`**: When using `and_then` with a callback that may return `Err`, define the initial `Result` with an explicit return type annotation. This ensures the Ok type is correctly inferred.
>
> ```python
> from pyropust import Ok, Err, Result
>
> def fetch_data() -> Result[int, str]: # Declare error type here
> def fetch_data() -> Result[int]: # Declare ok type here
> return Ok(42)
>
> def validate(x: int) -> Result[int, str]:
> def validate(x: int) -> Result[int]:
> return Err("invalid") if x < 0 else Ok(x)
>
> # Error type flows correctly through the chain
> result = fetch_data().and_then(validate)
> ```

#### Adding context and error codes

In real applications, errors often need additional context as they move up the stack.
pyropust provides helpers inspired by Rust’s `context` and error mapping patterns.

```python
from pyropust import Result, Err

def load_config(path: str) -> Result[str]:
return Err("file not found")

result = load_config("/etc/app.toml").context(
"failed to load application config",
code="config.load",
)
```

- `context(...)` adds human-readable context while preserving the original cause
- The original error is kept as a structured cause chain

You can also modify error codes for classification and observability:

```python
result = load_config("/etc/app.toml").with_code("config.not_found")

result = load_config("/etc/app.toml").map_err_code("startup")
```

Error codes are stable, machine-facing identifiers.
Messages are for humans and may change; codes are for branching, testing, and observability.

These helpers make it easy to:

- Add meaning at higher layers
- Classify failures without losing detail
- Keep error handling explicit and testable

### 2) Blueprint: typed pipelines

A **Blueprint** is a declarative pipeline that describes what happens to data, not how it is wired together.
Expand Down Expand Up @@ -223,12 +261,13 @@ The `@do` decorator enables linear, Rust-style propagation of `Result`.
from pyropust import Ok, Result, do

@do
def process(data: str) -> Result[str, object]:
def process(data: str) -> Result[str]:
text = yield Ok(data)
return Ok(text.upper())
```

When to use: `@do` reads like imperative code and is better when you need intermediate variables, early returns, or mixed steps.
Prefer `context` for adding meaning instead of catching exceptions.

This is not syntax sugar over exceptions — it is structured propagation of `Result` values.

Expand Down
105 changes: 71 additions & 34 deletions pyropust/__init__.pyi
Original file line number Diff line number Diff line change
Expand Up @@ -4,40 +4,52 @@ from typing import Generic, Never, TypeVar, overload

# Variance-sensitive type parameters require old-style TypeVar
T_co = TypeVar("T_co", covariant=True)
E_co = TypeVar("E_co", covariant=True)
In_contra = TypeVar("In_contra", contravariant=True)
Out_co = TypeVar("Out_co", covariant=True)

class Result(Generic[T_co, E_co]):
class Result(Generic[T_co]):
def is_ok(self) -> bool: ...
def is_err(self) -> bool: ...
def is_ok_and(self, predicate: Callable[[T_co], object]) -> bool: ...
def is_err_and(self, predicate: Callable[[E_co], object]) -> bool: ...
def is_err_and(self, predicate: Callable[[RopustError], object]) -> bool: ...
def unwrap(self) -> T_co: ...
def unwrap_err(self) -> E_co: ...
def unwrap_err(self) -> RopustError: ...
def expect(self, msg: str) -> T_co: ...
def expect_err(self, msg: str) -> E_co: ...
def expect_err(self, msg: str) -> RopustError: ...
def unwrap_or[U](self, default: U) -> T_co | U: ...
def unwrap_or_else[U](self, f: Callable[[E_co], U]) -> T_co | U: ...
def unwrap_or_else[U](self, f: Callable[[RopustError], U]) -> T_co | U: ...
def ok(self) -> Option[T_co]: ...
def err(self) -> Option[E_co]: ...
def map[U](self, f: Callable[[T_co], U]) -> Result[U, E_co]: ...
def map_err[U](self, f: Callable[[E_co], U]) -> Result[T_co, U]: ...
def err(self) -> Option[RopustError]: ...
def map[U](self, f: Callable[[T_co], U]) -> Result[U]: ...
def map_err(self, f: Callable[[RopustError], RopustError]) -> Result[T_co]: ...
def map_or[U](self, default: U, f: Callable[[T_co], U]) -> U: ...
def map_or_else[U](self, default_f: Callable[[E_co], U], f: Callable[[T_co], U]) -> U: ...
def inspect(self, f: Callable[[T_co], object]) -> Result[T_co, E_co]: ...
def inspect_err(self, f: Callable[[E_co], object]) -> Result[T_co, E_co]: ...
def and_[U](self, other: Result[U, E_co]) -> Result[U, E_co]: ...
def or_[F](self, other: Result[T_co, F]) -> Result[T_co, F]: ...
def or_else[F](self, f: Callable[[E_co], Result[T_co, F]]) -> Result[T_co, F]: ...
def and_then[U](self, f: Callable[[T_co], Result[U, E_co]]) -> Result[U, E_co]: ...
def flatten[T, E](self: Result[Result[T, E], E]) -> Result[T, E]: ...
def transpose[T, E](self: Result[Option[T], E]) -> Option[Result[T, E]]: ...
def map_or_else[U](
self, default_f: Callable[[RopustError], U], f: Callable[[T_co], U]
) -> U: ...
def inspect(self, f: Callable[[T_co], object]) -> Result[T_co]: ...
def inspect_err(self, f: Callable[[RopustError], object]) -> Result[T_co]: ...
def context(
self,
message: str,
*,
code: str = "context",
metadata: Mapping[str, str] | None = None,
op: str | None = None,
path: list[str | int] | None = None,
expected: str | None = None,
got: str | None = None,
) -> Result[T_co]: ...
def with_code(self, code: str) -> Result[T_co]: ...
def map_err_code(self, prefix: str) -> Result[T_co]: ...
def and_[U](self, other: Result[U]) -> Result[U]: ...
def or_(self, other: Result[T_co]) -> Result[T_co]: ...
def or_else(self, f: Callable[[RopustError], Result[T_co]]) -> Result[T_co]: ...
def and_then[U](self, f: Callable[[T_co], Result[U]]) -> Result[U]: ...
def flatten[T](self: Result[Result[T]]) -> Result[T]: ...
def transpose[T](self: Result[Option[T]]) -> Option[Result[T]]: ...
def unwrap_or_raise(self, exc: BaseException) -> T_co: ...
@classmethod
def attempt[T](
cls, f: Callable[[], T], *exceptions: type[BaseException]
) -> Result[T, RopustError]: ...
def attempt[T](cls, f: Callable[[], T], *exceptions: type[BaseException]) -> Result[T]: ...

class Option(Generic[T_co]):
def is_some(self) -> bool: ...
Expand All @@ -59,11 +71,11 @@ class Option(Generic[T_co]):
def or_else(self, f: Callable[[], Option[T_co]]) -> Option[T_co]: ...
def xor(self, other: Option[T_co]) -> Option[T_co]: ...
def flatten[T](self: Option[Option[T]]) -> Option[T]: ...
def transpose[T, E](self: Option[Result[T, E]]) -> Result[Option[T], E]: ...
def transpose[T](self: Option[Result[T]]) -> Result[Option[T]]: ...
def zip[U](self, other: Option[U]) -> Option[tuple[T_co, U]]: ...
def zip_with[U, R](self, other: Option[U], f: Callable[[T_co, U], R]) -> Option[R]: ...
def ok_or[E](self, error: E) -> Result[T_co, E]: ...
def ok_or_else[E](self, f: Callable[[], E]) -> Result[T_co, E]: ...
def ok_or(self, error: RopustError | BaseException | str) -> Result[T_co]: ...
def ok_or_else(self, f: Callable[[], RopustError | BaseException | str]) -> Result[T_co]: ...

class ErrorKind:
InvalidInput: ErrorKind
Expand Down Expand Up @@ -92,6 +104,33 @@ class RopustError:
def to_dict(self) -> dict[str, object]: ...
@classmethod
def from_dict(cls, data: dict[str, object]) -> RopustError: ...
@classmethod
def new(
cls,
code: str,
message: str,
*,
kind: ErrorKind | str | None = None,
op: str | None = None,
path: list[str | int] | None = None,
expected: str | None = None,
got: str | None = None,
metadata: Mapping[str, str] | None = None,
) -> RopustError: ...
@classmethod
def wrap(
cls,
err: BaseException | RopustError,
*,
code: str,
message: str,
kind: ErrorKind | str | None = None,
op: str | None = None,
path: list[str | int] | None = None,
expected: str | None = None,
got: str | None = None,
metadata: Mapping[str, str] | None = None,
) -> RopustError: ...

class Operator(Generic[In_contra, Out_co]): ...

Expand Down Expand Up @@ -229,17 +268,15 @@ class Op:
def to_uppercase() -> Operator[str, str]: ...
# END GENERATED OP

def Ok[T](value: T) -> Result[T, Never]: ...
def Err[E](error: E) -> Result[Never, E]: ...
def Ok[T](value: T) -> Result[T]: ...
def Err(error: RopustError | BaseException | str) -> Result[Never]: ...
def Some[T](value: T) -> Option[T]: ...
def None_() -> Option[Never]: ...

type Do[T, E, R] = Generator[Result[T, E], T, Result[R, E]]
type Do[T, R] = Generator[Result[T], T, Result[R]]

def do[**P, T, E, R](fn: Callable[P, Do[T, E, R]]) -> Callable[P, Result[R, E]]: ...
def run[OrigIn, Out](
blueprint: Blueprint[OrigIn, Out], input: OrigIn
) -> Result[Out, RopustError]: ...
def do[**P, T, R](fn: Callable[P, Do[T, R]]) -> Callable[P, Result[R]]: ...
def run[OrigIn, Out](blueprint: Blueprint[OrigIn, Out], input: OrigIn) -> Result[Out]: ...
def exception_to_ropust_error(exc: BaseException, code: str = "py_exception") -> RopustError: ...

# Overload 1: Decorator with exception types (@catch() or @catch(ValueError, ...))
Expand All @@ -250,14 +287,14 @@ def catch[R]( # type: ignore[overload-overlap]
exc: type[BaseException],
/,
*more_exceptions: type[BaseException],
) -> Callable[[Callable[..., R]], Callable[..., Result[R, RopustError]]]: ...
) -> Callable[[Callable[..., R]], Callable[..., Result[R]]]: ...

# Overload 2: Decorator without arguments (@catch())
@overload
def catch[R]() -> Callable[[Callable[..., R]], Callable[..., Result[R, RopustError]]]: ...
def catch[R]() -> Callable[[Callable[..., R]], Callable[..., Result[R]]]: ...

# Overload 3: Bare decorator usage (@catch)
@overload
def catch[R](fn: Callable[..., R], /) -> Callable[..., Result[R, RopustError]]: ...
def catch[R](fn: Callable[..., R], /) -> Callable[..., Result[R]]: ...

__all__: list[str]
17 changes: 7 additions & 10 deletions pyropust/catch.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,7 @@
from functools import wraps
from typing import overload

from .pyropust_native import Result, RopustError
from .pyropust_native import Result


def _is_exception_type(value: object) -> bool:
Expand All @@ -14,32 +14,29 @@ def _is_exception_type(value: object) -> bool:
def _decorate[**P, R](
fn: Callable[P, R],
exc_types: tuple[type[BaseException], ...],
) -> Callable[P, Result[R, RopustError]]:
) -> Callable[P, Result[R]]:
@wraps(fn)
def wrapper(*args: P.args, **kwargs: P.kwargs) -> Result[R, RopustError]:
def wrapper(*args: P.args, **kwargs: P.kwargs) -> Result[R]:
return Result.attempt(lambda: fn(*args, **kwargs), *exc_types)

return wrapper


# Overload 1: Bare decorator usage (@catch)
@overload
def catch[**P, R](fn: Callable[P, R], /) -> Callable[P, Result[R, RopustError]]: ...
def catch[**P, R](fn: Callable[P, R], /) -> Callable[P, Result[R]]: ...


# Overload 2: Decorator with exception types (@catch() or @catch(ValueError))
@overload
def catch[**P, R](
*exc_types: type[BaseException],
) -> Callable[[Callable[P, R]], Callable[P, Result[R, RopustError]]]: ...
) -> Callable[[Callable[P, R]], Callable[P, Result[R]]]: ...


def catch[**P, R](
*args: type[BaseException] | Callable[P, R],
) -> (
Callable[P, Result[R, RopustError]]
| Callable[[Callable[P, R]], Callable[P, Result[R, RopustError]]]
):
) -> Callable[P, Result[R]] | Callable[[Callable[P, R]], Callable[P, Result[R]]]:
"""Convert exceptions into Result using RopustError.

Can be used as @catch or @catch(ValueError, TypeError).
Expand All @@ -59,7 +56,7 @@ def catch[**P, R](
if not exc_types:
exc_types = [Exception]

def decorator(fn: Callable[P, R]) -> Callable[P, Result[R, RopustError]]:
def decorator(fn: Callable[P, R]) -> Callable[P, Result[R]]:
return _decorate(fn, tuple(exc_types))

return decorator
12 changes: 6 additions & 6 deletions pyropust/do.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,11 +7,11 @@
from .pyropust_native import Result


def do[**P, T, E, R](
fn: Callable[P, Generator[Result[T, E], T, Result[R, E]]],
) -> Callable[P, Result[R, E]]:
def do[**P, T, R](
fn: Callable[P, Generator[Result[T], T, Result[R]]],
) -> Callable[P, Result[R]]:
@wraps(fn)
def wrapper(*args: P.args, **kwargs: P.kwargs) -> Result[R, E]:
def wrapper(*args: P.args, **kwargs: P.kwargs) -> Result[R]:
gen = fn(*args, **kwargs)
if not hasattr(gen, "send"):
raise TypeError("@do function must return a generator")
Expand All @@ -28,9 +28,9 @@ def wrapper(*args: P.args, **kwargs: P.kwargs) -> Result[R, E]:
if not isinstance(current, Result):
raise TypeError("yielded value must be Result")
if current.is_err():
# On error, the value type is irrelevant, so cast to Result[R, E]
# On error, the value type is irrelevant, so cast to Result[R]
if TYPE_CHECKING:
return cast("Result[R, E]", current)
return cast("Result[R]", current)
return current
try:
current = gen.send(current.unwrap())
Expand Down
Loading