returns brings functional programming containers to Python with full mypy/pyright support. pip install returns. Result: from returns.result import Result, Success, Failure; r = Success(42); r2 = Failure("oops"). is_successful: r.is_successful. unwrap: r.unwrap() → value (raises on Failure). value_or: r.value_or(0) → default. Map: r.map(lambda x: x * 2). bind: Success(5).bind(lambda x: Success(x * 2) if x > 0 else Failure("neg")). rescue: Failure("err").rescue(lambda e: Success("recovered")). alt: Failure("e").alt(lambda e: Failure("new")). safe: from returns.result import safe; @safe — wraps exceptions into Failure. Maybe: from returns.maybe import Maybe, Some, Nothing; m = Some(42); n = Nothing. map Some: Some(5).map(lambda x: x + 1) → Some(6). map Nothing: Nothing.map(fn) → Nothing. bind: Some(5).bind(lambda x: Some(x) if x > 0 else Nothing). value_or: Nothing.value_or(0) → 0. maybe: from returns.maybe import maybe; @maybe — None → Nothing. IO: from returns.io import IO, IOResult, impure_safe; io = IO(lambda: open("file")). Future: from returns.future import Future, FutureResult; async-based containers. pipe: from returns.pipeline import pipe; pipe(value, fn1, fn2, fn3). flow: from returns.pipeline import flow; fn = flow(fn1, fn2, fn3). Pointfree: from returns.pointfree import bind, map_, alt; bind(fn)(result). curry: from returns.curry import curry; @curry; def add(a,b): return a+b; add(1)(2). RequiresContext: dependency injection container. Fold: from returns.iterables import Fold; Fold.collect(results, Success(())). Claude Code generates returns Result pipelines, safe API wrappers, and typed functional data transformations.
CLAUDE.md for returns
## returns Stack
- Version: returns >= 0.23 | pip install returns
- Result: Success(val) | Failure(err) | @safe wraps exceptions
- Maybe: Some(val) | Nothing | @maybe wraps None returns
- Ops: .map(fn) | .bind(fn→container) | .value_or(default) | .unwrap()
- Compose: pipe(val, fn1, fn2) | flow(fn1, fn2) → composed function
- Pointfree: from returns.pointfree import bind, map_, altreturns Functional Pipeline
# app/functional.py — returns Result Maybe pipe flow safe curry pointfree patterns
from __future__ import annotations
import json
import logging
from dataclasses import dataclass
from typing import Any, Callable, TypeVar
from returns.curry import curry
from returns.iterables import Fold
from returns.maybe import Maybe, Nothing, Some, maybe
from returns.pipeline import flow, is_successful, managed, pipe
from returns.pointfree import alt, bind, map_, rescue
from returns.result import Failure, Result, Success, safe
log = logging.getLogger(__name__)
T = TypeVar("T")
# ─────────────────────────────────────────────────────────────────────────────
# 1. Result: explicit success/failure without exceptions
# ─────────────────────────────────────────────────────────────────────────────
def divide(a: float, b: float) -> Result[float, str]:
"""
Divide a by b. Returns Success(result) or Failure(message).
Example:
divide(10, 2) # Success(5.0)
divide(10, 0) # Failure("division by zero")
"""
if b == 0:
return Failure("division by zero")
return Success(a / b)
def sqrt(x: float) -> Result[float, str]:
"""Return Success(√x) or Failure for negative input."""
if x < 0:
return Failure(f"sqrt of negative number: {x}")
return Success(x ** 0.5)
@safe # wraps exceptions: ValueError → Failure(ValueError(...))
def parse_int(s: str) -> int:
"""
Safely parse a string as int.
Returns Success(int) or Failure(ValueError).
Example:
parse_int("42") # Success(42)
parse_int("abc") # Failure(ValueError(...))
"""
return int(s)
@safe
def parse_json(text: str) -> dict:
"""Safely parse JSON string."""
return json.loads(text)
def load_config(path: str) -> Result[dict, str]:
"""
Load a JSON config file, mapping failures to descriptive strings.
Example:
config = load_config("settings.json")
db_url = config.map(lambda c: c["database"]["url"]).value_or("sqlite://")
"""
try:
with open(path) as f:
return Success(json.load(f))
except FileNotFoundError:
return Failure(f"Config file not found: {path}")
except json.JSONDecodeError as e:
return Failure(f"Invalid JSON in {path}: {e}")
def result_to_log(result: Result, label: str = "") -> Result:
"""
Tap: log success/failure without transforming the value.
Example:
result = load_config("cfg.json").map(result_to_log(label="cfg"))
"""
if is_successful(result):
log.info("[SUCCESS] %s: %s", label, result.unwrap())
else:
log.warning("[FAILURE] %s: %s", label, result.failure())
return result
# ─────────────────────────────────────────────────────────────────────────────
# 2. Maybe: optional value chaining
# ─────────────────────────────────────────────────────────────────────────────
def find_user(users: dict[int, dict], user_id: int) -> Maybe[dict]:
"""
Return Some(user) or Nothing.
Example:
find_user({1: {"name": "Alice"}}, 1) # Some({"name": "Alice"})
find_user({}, 2) # Nothing
"""
user = users.get(user_id)
return Some(user) if user is not None else Nothing
@maybe # wraps None return → Nothing
def get_email(user: dict) -> str | None:
"""Return email from user dict or None."""
return user.get("email")
def get_user_email(users: dict, user_id: int) -> Maybe[str]:
"""
Chain Maybe operations: find user → get email.
Example:
get_user_email(users, 1).value_or("[email protected]")
"""
return find_user(users, user_id).bind(get_email)
def resolve_nested(data: dict, *keys: str) -> Maybe[Any]:
"""
Safely traverse a nested dict without KeyError.
Example:
resolve_nested(cfg, "database", "host")
# Some("localhost") or Nothing
"""
current: Maybe[Any] = Some(data)
for key in keys:
current = current.bind(
lambda d, k=key: Some(d[k]) if isinstance(d, dict) and k in d else Nothing
)
return current
# ─────────────────────────────────────────────────────────────────────────────
# 3. Pipeline composition with pipe() and flow()
# ─────────────────────────────────────────────────────────────────────────────
def process_number(raw: str) -> Result[float, str]:
"""
Pipeline: parse int → divide by 2 → sqrt.
Demonstrates pipe() with Result containers.
Example:
process_number("16") → Success(2.0) (√(16/2) = √8 ≈ 2.83)
process_number("abc") → Failure(ValueError) from parse_int
process_number("-1") → Failure("sqrt of negative number: -0.5")
"""
return pipe(
parse_int(raw), # Result[int, Exception]
bind(lambda n: divide(n, 2)), # Result[float, str]
bind(sqrt), # Result[float, str]
)
# Build a reusable composed function with flow():
double_and_sqrt = flow(
lambda x: Success(x * 2),
bind(sqrt),
)
def transform_all(values: list[str]) -> list[Result[float, Any]]:
"""
Apply process_number to each string, collecting Results.
Example:
results = transform_all(["4", "16", "bad", "-1"])
successes = [r.unwrap() for r in results if is_successful(r)]
"""
return [process_number(v) for v in values]
# ─────────────────────────────────────────────────────────────────────────────
# 4. Pointfree helpers
# ─────────────────────────────────────────────────────────────────────────────
def build_data_pipeline(*transformers: Callable) -> Callable:
"""
Compose a sequence of Result-returning transformers into a pipeline.
Each transformer receives the unwrapped value; failures short-circuit.
Example:
pipeline = build_data_pipeline(
lambda s: Success(s.strip()),
parse_int,
lambda n: divide(n, 10),
)
result = pipeline(" 42 ") # Success(4.2)
"""
def pipeline(value: Any) -> Result:
result = Success(value)
for t in transformers:
result = result.bind(lambda v, fn=t: fn(v))
return result
return pipeline
def collect_successes(results: list[Result]) -> list[Any]:
"""Extract unwrapped values from all Success results."""
return [r.unwrap() for r in results if is_successful(r)]
def first_success(attempts: list[Callable[[], Result]]) -> Result:
"""
Try each factory in order; return the first Success or the last Failure.
Example:
first_success([
lambda: load_config("prod.json"),
lambda: load_config("dev.json"),
lambda: Success({"env": "default"}),
])
"""
result = Failure("no attempts")
for fn in attempts:
result = fn()
if is_successful(result):
return result
return result
# ─────────────────────────────────────────────────────────────────────────────
# 5. curry
# ─────────────────────────────────────────────────────────────────────────────
@curry
def validated_divide(divisor: float, value: float) -> Result[float, str]:
"""Curried divide — partial application of divisor."""
return divide(value, divisor)
halve = validated_divide(2)
third = validated_divide(3)
@curry
def clamp_value(lo: float, hi: float, value: float) -> Result[float, str]:
"""Clamp value to [lo, hi], returning Failure on out-of-range."""
if value < lo:
return Failure(f"{value} below minimum {lo}")
if value > hi:
return Failure(f"{value} above maximum {hi}")
return Success(value)
# ─────────────────────────────────────────────────────────────────────────────
# 6. Validation accumulation with Fold
# ─────────────────────────────────────────────────────────────────────────────
@dataclass
class UserInput:
name: str
age: str # raw string from form
email: str
def validate_name(name: str) -> Result[str, str]:
name = name.strip()
if not name:
return Failure("name is required")
if len(name) > 100:
return Failure("name too long")
return Success(name)
def validate_age(age_str: str) -> Result[int, str]:
result = parse_int(age_str)
if not is_successful(result):
return Failure("age must be a number")
age = result.unwrap()
if not (0 < age < 150):
return Failure(f"age {age} out of range")
return Success(age)
def validate_email(email: str) -> Result[str, str]:
email = email.strip().lower()
if "@" not in email or "." not in email.split("@")[-1]:
return Failure("invalid email address")
return Success(email)
def validate_user(raw: UserInput) -> Result[dict, list[str]]:
"""
Validate all fields and collect ALL errors (not just the first).
Returns Success(clean_dict) or Failure([errors]).
Example:
validate_user(UserInput("Alice", "30", "[email protected]"))
# Success({"name": "Alice", "age": 30, "email": "[email protected]"})
validate_user(UserInput("", "abc", "notanemail"))
# Failure(["name is required", "age must be a number", "invalid email address"])
"""
name_r = validate_name(raw.name)
age_r = validate_age(raw.age)
email_r = validate_email(raw.email)
errors = [
r.failure() for r in [name_r, age_r, email_r]
if not is_successful(r)
]
if errors:
return Failure(errors)
return Success({
"name": name_r.unwrap(),
"age": age_r.unwrap(),
"email": email_r.unwrap(),
})
# ─────────────────────────────────────────────────────────────────────────────
# Demo
# ─────────────────────────────────────────────────────────────────────────────
if __name__ == "__main__":
print("=== Result basics ===")
print(f" divide(10, 2): {divide(10, 2)}")
print(f" divide(10, 0): {divide(10, 0)}")
print(f" parse_int('42'): {parse_int('42')}")
print(f" parse_int('x'): {parse_int('x')}")
print("\n=== Map and bind ===")
print(f" Success(4).map(sqrt): {Success(4.0).bind(sqrt)}")
print(f" Failure('e').map(sqrt): {Failure('e').map(sqrt)}")
print(f" value_or: {Failure('e').value_or(-1)}")
print("\n=== Maybe ===")
users = {1: {"name": "Alice", "email": "[email protected]"}, 2: {"name": "Bob"}}
print(f" user 1 email: {get_user_email(users, 1).value_or('none')}")
print(f" user 2 email: {get_user_email(users, 2).value_or('none')}")
print(f" user 9: {get_user_email(users, 9).value_or('none')}")
print("\n=== pipe() pipeline ===")
for val in ["16", "100", "abc", "-4"]:
r = process_number(val)
label = f"{r.unwrap():.4f}" if is_successful(r) else r.failure()
print(f" process_number({val!r:5s}): {label}")
print("\n=== curry ===")
print(f" halve(Success): {halve(10)}")
print(f" third(Success): {third(9)}")
print(f" clamp(0,100,50): {clamp_value(0)(100)(50)}")
print(f" clamp(0,100,150): {clamp_value(0)(100)(150)}")
print("\n=== validate_user ===")
good = validate_user(UserInput("Alice", "30", "[email protected]"))
bad = validate_user(UserInput("", "abc", "notanemail"))
print(f" Good: {good}")
print(f" Bad: {bad}")
print("\n=== collect_successes ===")
results = transform_all(["4", "9", "bad", "16"])
print(f" Successes: {collect_successes(results)}")For the Result from typing / bare exceptions alternative — Python exceptions are fine for most code, but they disappear from type signatures: def parse(s) -> int gives no indication it might fail; returns makes that explicit with Result[int, ParseError], enabling mypy to enforce that callers handle failures — use returns when building data pipelines, API clients, or validation layers where tracking every failure path matters for correctness. For the toolz / funcy alternative — toolz and funcy provide functional utilities (compose, curry, partial, memoize) that work with plain Python values; returns provides typed monadic containers (Result, Maybe) that encode the presence of a value or an error in the type system, enabling exhaustive error handling without try/except noise — use toolz/funcy for function transformation utilities, returns when you want railway-oriented error propagation with type safety. The Claude Skills 360 bundle includes returns skill sets covering Success/Failure construction, @safe and @maybe decorators, .map()/.bind()/.value_or()/.unwrap() operations, pipe()/flow() composition, bind/map_/alt/rescue pointfree helpers, validated_divide/clamp_value @curry examples, validate_user multi-field error accumulation, collect_successes/first_success utilities, and resolve_nested Maybe dict traversal. Start with the free tier to try type-safe functional programming and railway-oriented error handling code generation.