typeric 0.4.0

Utility types for Python, such as `Result`, `Option`, etc.

📦 typeric

typeric is a practical type utility toolkit for Python, focused on clarity, safety, and ergonomics. It was originally built to make my own development experience smoother, but I hope it proves useful to others as well.
It currently provides lightweight, pattern-matchable types like Result and Option — inspired by Rust — with plans to include more common type patterns and error-handling abstractions.

pip install typeric

---

🚀 Features

• 🎯 Rust-style error propagation (the ? operator)
  .spread() + @spreadable short-circuits a function on the first Err/NONE, exactly like Rust's ? — the core of this library. Works for both Result and Option, sync and async.

• ✅ Functional-style Result type
  Ok(value) and Err(error) with powerful .map(), .and_then(), .combine(), .spread() helpers — inspired by Rust’s Result.

• 🌀 Lightweight Option type
  Some(value) and NONE to handle nullable data safely, with .map(), .unwrap_or(), .is_some() and more.

• 🔁 Seamless conversion decorators

  • @resulty: Wraps any function to return Result instead of raising exceptions.
  • @resulty(catch=...): Catch only specific exception types and keep the exception object as the error.
  • @optiony: Wraps any function to return Option, converting None or exceptions into NONE.
  • @optiony(catch=...): Only the listed exception types become NONE; everything else propagates.

• 🔀 Result ↔ Option interop
  result.to_option(), option.ok_or(error) / option.ok_or_else(f) — bridge the two worlds and keep one propagation chain.

• 📚 Collection helpers
  collect_results, partition_results, gather_results (async), combine_results (flat combine), collect_options — batch up fallible operations without hand-written loops.

• 🔎 Static narrowing guards
  is_ok / is_err / is_some / is_none are TypeIs guards: after if is_ok(res):, type checkers know res is Ok[T].

• 🧰 Resilience decorators
  @retry (retries on Err or listed exceptions with backoff), @timeout_async (expiry becomes Err(TimeoutError)), @cache_ok (memoizes only successes — failures are never cached), @log_err (failure branches logged automatically). All propagation-safe.

• 🧩 Pattern matching support
  Supports Python’s match syntax via __match_args__ for both Ok/Err and Some/NONE.

• 🔒 Immutable and composable
  Safe and clean method chains using .map(), .combine(), .inspect(), etc.

• 🔧 Clean type signatures
  Fully typed: Result[T, E] and Option[T] with static analysis and IDE support.

• 🛠️ Extensible foundation
  Designed for easy extension — more algebraic types (Either, Validated, etc.) can be added naturally.

---

🔍 Quick Example

Error propagation — Python's ? operator

The essence of Rust's error handling is not Result itself, but the ? operator: fail fast, propagate the error, keep the happy path flat. typeric brings this to Python with .spread() and @spreadable:

from typeric import Ok, Err, Result, spreadable

def find_user(uid: int) -> Result[str, str]:
    return Ok("alice") if uid == 1 else Err("user not found")

def check_permission(user: str) -> Result[str, str]:
    return Ok(user) if user == "alice" else Err("permission denied")

@spreadable
def handle_request(uid: int) -> Result[str, str]:
    user = find_user(uid).spread()          # like `find_user(uid)?` in Rust
    granted = check_permission(user).spread()
    return Ok(f"hello, {granted}")

handle_request(1)   # Ok('hello, alice')
handle_request(42)  # Err('user not found') — propagated, no if/else ladder

Option propagates the same way with @spreadable_option:

from typeric import Some, NONE, Option, spreadable_option

def maybe_positive(x: int) -> Option[int]:
    return Some(x) if x > 0 else NONE

@spreadable_option
def add_one(x: int) -> Option[int]:
    val = maybe_positive(x).spread()   # NONE short-circuits the whole function
    return Some(val + 1)

add_one(5)    # Some(6)
add_one(-1)   # NONE

Async versions: @spreadable_async and @spreadable_option_async.

Result

from typeric import Result, Ok, Err, resulty, resulty_async, spreadable

def parse_number(text: str) -> Result[int, str]:
    try:
        return Ok(int(text))
    except ValueError:
        return Err("Not a number")

match parse_number("42"):
    case Ok(value):
        print("Parsed:", value)
    case Err(error):
        print("Failed:", error)

# let function return Result[T,str]
@resulty
def add(x: int, y: int) -> int:
    return x + y

# or catch only specific exceptions and keep the exception object
@resulty(catch=(ValueError, KeyError))
def parse_strict(raw: str) -> int:   # -> Result[int, ValueError | KeyError]
    return int(raw)

res = add(1, 2)
if res.is_ok():
    print("Result:", res.unwrap())
else:
    print("Error:", res.err)

# let async function return Result[T,str]
@resulty_async
async def async_add(x: int, y: int) -> int:
    return x + y

res = await async_add(1, 2)
if res.is_ok():
    print("Result:", res.unwrap())
else:
    print("Error:", res.err)

def func_a(x: int) -> Result[int, str]:
    if x < 0:
        return Err("negative input")
    return Ok(x * 2)


@spreadable
def func_b(y: int) -> Result[int, str]:
    a = func_a(y).spread()
    return Ok(a + 1)


def test_func_b_success():
    assert func_b(5) == Ok(11)  # 5*2=10 +1=11


def test_func_b_propagate_error():
    assert func_b(-2) == Err("negative input")

def validate_username(username: str) -> Result[str, str]:
    if username.strip():
        return Ok(username)
    return Err("Username is empty")


def validate_age(age: int) -> Result[int, str]:
    if age > 0:
        return Ok(age)
    return Err("Age must > 0")


def validate_email(email: str) -> Result[str, str]:
    if "@" in email:
        return Ok(email)
    return Err("Invalid email")


# ✅ results combine
def validate_user_data(
    username: str, age: int, email: str
) -> Result[tuple[tuple[str, int], str], str]:
    return (
        validate_username(username)
        .combine(validate_age(age))
        .combine(validate_email(email))
    )


result1 = validate_user_data("alice", 30, "[e-mail]")
print(result1)  # Ok((('alice', 30), '[e-mail]'))

result2 = validate_user_data("", -5, "invalid-email")
print(result2.errs)  # ['Username is empty', 'Age must > 0', 'Invalid email']

Option

from typeric import Option, Some, NONE, NoneType, optiony, optiony_async
from typeric.wrap_funcs import get_time_sync

def maybe_get(index: int, items: list[str]) -> Option[str]:
    if 0 <= index < len(items):
        return Some(items[index])
    return NONE

match maybe_get(1, ["a", "b", "c"]):
    case Some(value):
        print("Got:", value)
    case NoneType():  # note: `case NONE:` would be a capture pattern matching anything
        print("Nothing found")

@get_time_sync # This decorator is used for synchronous functions to measure execution time.
@optiony
def get_number(x: int) -> int | None:
    if x > 0:
        return x
    return None

@optiony_async
async def fetch_data(flag: bool) -> str | None:
    if flag:
        return "data"
    return None

---

📖 API Reference

Imports

Everything is available from the top-level package:

from typeric import (
    # Result
    Result, Ok, Err,
    # Option
    Option, Some, NONE, NoneType,
    # conversion decorators
    resulty, resulty_async, optiony, optiony_async,
    # propagation decorators (Rust's `?`)
    spreadable, spreadable_async, spreadable_option, spreadable_option_async,
    # collection helpers
    collect_results, partition_results, gather_results, combine_results, collect_options,
    # TypeIs narrowing guards
    is_ok, is_err, is_some, is_none,
    # resilience decorators
    retry, retry_async, timeout_async, cache_ok, cache_ok_async, log_err, log_err_async,
    # errors & helpers
    UnwrapError, NoneTypeError, EarlyReturn, AggregatedErrors,
)

Module paths also work: typeric.result, typeric.option, typeric.wrap_funcs.

Result[T, E] = Ok[T] | Err[E]

Method	Ok(v)	Err(e)
.is_ok() / .is_err()	True / False	False / True
.ok	v	—
.err	—	e
.errs	None	all aggregated errors as list[E]
.unwrap()	v	raises e if it is an exception, else UnwrapError
.unwrap_or(default)	v	default
.unwrap_or_else(f)	v	f(e)
.unwrap_err()	raises UnwrapError	e
.expect(msg)	v	raises UnwrapError(f"{msg}: ...")
.map(f)	Ok(f(v))	unchanged
.map_err(f)	unchanged	Err(f(e)), mapping all aggregated errors
.map_or(default, f)	f(v)	default
.and_then(f)	f(v) (returns a Result)	unchanged
.or_else(f)	unchanged	f(e) (returns a Result)
.flatten()	Ok(Ok(v)) → Ok(v)	unchanged
.inspect(f) / .inspect_err(f)	side effect on v / no-op	no-op / side effect on e
.combine(other)	pair up values: Ok((v, other_v))	accumulate errors into .errs
.to_option()	Some(v)	NONE
.spread()	v	short-circuit: propagate to nearest @spreadable

Notes:

• Values are immutable — combine, map_err, etc. always return new instances and never mutate the operands, so Result values can be safely reused and shared.
• Ok/Err (and Some/NONE) are hashable and usable in sets/dicts as long as the wrapped value is hashable.
• combine accumulates: chain it across validations, then read every failure from .errs (first failure stays in .err).

Option[T] = Some[T] | NoneType

NONE is the singleton NoneType() instance.

Method	Some(v)	NONE
.is_some() / .is_none()	True / False	False / True
.unwrap()	v	raises NoneTypeError
.unwrap_or(default)	v	default
.unwrap_or_else(f)	v	f()
.expect(msg)	v	raises NoneTypeError(msg)
.map(f)	Some(f(v))	NONE
.and_then(f)	f(v) (returns an Option)	NONE
.filter(pred)	Some(v) if pred(v) else NONE	NONE
.zip(other)	Some((v, other_v)) if both Some	NONE
.ok_or(error) / .ok_or_else(f)	Ok(v)	Err(error) / Err(f())
.spread()	v	short-circuit: propagate to nearest @spreadable_option

Pattern matching: use case Some(value): and case NoneType():. (case NONE: is a capture pattern in Python — it matches anything and rebinds the name.)

Decorators

Decorator	Applies to	Behavior
@resulty	sync def	returns Ok(value); raised exceptions become Err(str(e)); a returned Result passes through with errors stringified
@resulty(catch=T | (T, ...))	sync def	catches only the listed exception types and keeps the exception object: Result[T, ValueError]; other exceptions propagate; returned Results pass through untouched
@resulty_async / @resulty_async(catch=...)	async def	same as the sync versions
@optiony	sync def	returns Some(value); None or a raised exception becomes NONE; a returned Option passes through
@optiony(catch=T | (T, ...))	sync def	only the listed exception types become NONE; other exceptions propagate
@optiony_async / @optiony_async(catch=...)	async def	same as the sync versions
@spreadable	sync def returning Result	catches .spread() short-circuits and returns them as Err
@spreadable_async	async def returning Result	same as @spreadable
@spreadable_option	sync def returning Option	catches .spread() short-circuits and returns NONE
@spreadable_option_async	async def returning Option	same as @spreadable_option

Propagation rules:

• .spread() works by raising an internal EarlyReturn exception, caught by the nearest spreadable* wrapper — the function using .spread() must be decorated, otherwise EarlyReturn escapes to the caller.
• @resulty / @optiony deliberately let EarlyReturn pass through, so they stack safely with the propagation decorators:

@spreadable      # catches the propagation
@resulty         # converts raised exceptions / bare returns
def handler(uid: int) -> int:
    user = find_user(uid).spread()   # Err propagates out past @resulty, intact
    return len(user)

Collection helpers

from typeric import (
    collect_results, partition_results, gather_results, combine_results, collect_options,
)

results = [parse(x) for x in ["1", "2", "oops"]]

collect_results(results)     # Result[list[T], E] — fail-fast, first Err wins
partition_results(results)   # (list[T], list[E]) — run everything, split values/errors
await gather_results(fetch(1), fetch(2))   # run coroutines concurrently, then collect

# flat combine: no nested tuples, errors aggregate into .errs
combine_results(Ok("alice"), Ok(30), Ok("a@b.c"))   # Ok(('alice', 30, 'a@b.c'))
combine_results(Ok("alice"), Err("bad age"), Err("bad email")).errs
# ['bad age', 'bad email']

collect_options([Some(1), Some(2)])   # Some([1, 2]); any NONE -> NONE

Static narrowing — TypeIs guards

res.is_ok() returns a plain bool, which type checkers cannot narrow on. The module-level guards can:

from typeric import is_ok

res: Result[int, str] = parse("42")
if is_ok(res):
    reveal_type(res)   # Ok[int] — .ok / .unwrap() fully typed
else:
    reveal_type(res)   # Err[str]

Also available: is_err, is_some, is_none.

Result ↔ Option interop

parse("42").to_option()          # Result -> Option: Ok -> Some, Err -> NONE
maybe_user.ok_or("not found")    # Option -> Result: Some -> Ok, NONE -> Err("not found")

@spreadable
def load_config(path: str) -> Result[str, str]:
    raw = read_file(path).spread()                       # Result chain
    key = find_key(raw).ok_or("key missing").spread()    # Option joins the same chain
    return Ok(key)

Resilience decorators — typeric.decorators

Decorator	Applies to	Behavior
@retry(times=3, on=(), backoff=0.0, factor=2.0)	sync def	re-calls while the function returns Err; by default exceptions are not retried (they propagate immediately) — opt in with on=(ConnectionError, ...) for failures known to be transient. Sleeps backoff * factor**n between attempts; the final Err is returned / final exception re-raised unchanged. NONE is a value, not a failure — never retried
@retry_async(...)	async def	same, sleeping with asyncio.sleep
@timeout_async(seconds)	async def	deadline expiry returns Err(TimeoutError); a TimeoutError raised by the function body itself (e.g. an OS-level socket timeout) is not the deadline and propagates. Result/Option returns pass through untouched, bare values become Ok(value)
@cache_ok(maxsize=128)	sync def	LRU-memoizes only Ok/Some results; Err, NONE, and exceptions are never cached, so transient failures retry naturally. Like lru_cache, hits return the same instance — treat cached values as immutable. Exposes .cache_clear()
@cache_ok_async(maxsize=128)	async def	same (concurrent misses may compute twice)
@log_err / @log_err(level=...)	sync def	logs Err/NONE returns on the "typeric" logger, returns them unchanged; Ok/Some stay silent
@log_err_async	async def	same

All of them let EarlyReturn pass through, so .spread() propagation keeps working inside a retried/cached/logged function.

from typeric import Ok, Err, Result, retry_async, timeout_async, cache_ok_async, log_err_async

@cache_ok_async()          # outermost: cache the final success
@log_err_async             # log what failed after retries gave up
@retry_async(times=3, backoff=0.5)   # retry transient failures
@timeout_async(2.0)        # innermost: each attempt is bounded
async def fetch_profile(uid: int) -> Result[dict, str]:
    ...

Recommended stacking order: cache_ok outermost, then log_err, then retry, with timeout_async innermost so every retry attempt gets its own time budget.

Timing decorators — typeric.wrap_funcs

@get_time_sync / @get_time_async log call arguments and execution time via the standard-library logger named "typeric" (no third-party dependency). Enable output with:

import logging
logging.basicConfig(level=logging.INFO)

---

✅ Test

Run tests with:

uv run pytest -v

---

📦 Roadmap

• Async support (resulty_async, optiony_async, spreadable_async, spreadable_option_async)
• Rust-style ? propagation for both Result and Option
• Result ↔ Option interop (to_option, ok_or, ok_or_else)
• Collection helpers (collect_results, partition_results, gather_results, combine_results, collect_options)
• Exception-preserving decorators (@resulty(catch=...), @optiony(catch=...))
• TypeIs narrowing guards (is_ok, is_err, is_some, is_none)
• Resilience decorators (retry, timeout_async, cache_ok, log_err)
• Try, Either, NonEmptyList, etc.

---

📄 License

MIT