A Python package for elegant error handling, inspired by Rust's Result type.
pip install safe-resultsafe-result provides type-safe objects that represent either success (Ok) or failure (Err). This approach enables more explicit error handling without relying on try/catch blocks, making your code more predictable and easier to reason about.
Key features:
- 100% test coverage
- Type-safe result handling with full generics support
- Pattern matching support for elegant error handling
- Type guards for safe access and type narrowing
- Decorators to automatically wrap function returns in
Resultobjects - Methods for transforming and chaining results (
map,map_async,and_then,and_then_async,flatten) - Methods for accessing values, providing defaults or propagating errors within a
@safecontext - Handy traceback capture for comprehensive error information
Create Result objects directly or use the provided decorators.
from safe_result import Err, Ok, Result, ok
def divide(a: int, b: int) -> Result[float, ZeroDivisionError]:
if b == 0:
return Err(ZeroDivisionError("Cannot divide by zero")) # Failure case
return Ok(a / b) # Success case
# Function signature clearly communicates potential failure modes
foo = divide(10, 0) # -> Result[float, ZeroDivisionError]
# Type checking will prevent unsafe access to the value
bar = 1 + foo.value
# ^^^^^^^^^ Type checker indicates the error:
# "Operator '+' not supported for types 'Literal[1]' and 'float | None'"
# Safe access pattern using the type guard function
if ok(foo): # Verifies foo is an Ok result and enables type narrowing
bar = 1 + foo.value # Safe! Type checker knows the value is a float here
else:
# Handle error case with full type information about the error
print(f"Error: {foo.error}")
# Pattern matching is also a great way to handle results
match foo:
case Ok(value):
print(f"Success: {value}")
case Err(ZeroDivisionError() as e):
print(f"Division Error: {e}")Decorators simplify wrapping existing functions.
@safe: Catches any Exception and returns Result[ReturnType, Exception].
from safe_result import ok, safe
@safe
def may_fail(data: str) -> int:
return int(data)
def do_something():
result1 = may_fail("123") # -> Ok(123)
result2 = may_fail("abc") # -> Err(ValueError("invalid literal for int() with base 10: 'abc'"))
if ok(result1):
do_something_else(result1.value)
else:
print(f"Caught error: {result1.error}")
return result1
# Or even better by inverting the condition
if not ok(result2):
print(f"Caught error: {result2.error}")
return result2
# Continue with the rest of the function
do_something_else(result2.value)@safe_with(*ExceptionTypes): Catches only the specified exception types, returning Result[ReturnType, Union[ExceptionTypes]]. Other exceptions are raised normally.
from typing import Any
from safe_result import err_type, safe_with
@safe_with(ValueError, TypeError)
def process_input(data: Any) -> str:
if not isinstance(data, str):
raise TypeError("Input must be a string")
if not data:
raise ValueError("Input cannot be empty")
return f"Processed: {data}"
res1 = process_input("hello") # -> Ok('Processed: hello')
res2 = process_input("") # -> Err(ValueError('Input cannot be empty'))
res3 = process_input(123) # -> Err(TypeError('Input must be a string'))
res4 = process_input(None) # -> Raises TypeError (caught by decorator)
# Use err_type for specific error handling
if err_type(res2, ValueError):
print("ValueError occurred!")@safe_async and @safe_async_with work identically for asynchronous functions. asyncio.CancelledError is never caught and always re-raised.
import asyncio
from safe_result import Err, Ok, safe_async, safe_async_with
@safe_async
async def fetch_data(url: str) -> str:
await asyncio.sleep(0.1) # Simulate network
if "invalid" in url:
raise ValueError("Invalid URL")
return f"Data from {url}"
@safe_async_with(ConnectionError)
async def fetch_specific(url: str) -> str:
await asyncio.sleep(0.1)
if "timeout" in url:
raise ConnectionError("Timeout")
return f"Data from {url}"
async def main():
result1 = await fetch_data("valid-url") # -> Ok('Data from valid-url')
result2 = await fetch_data("invalid-url") # -> Err(ValueError('Invalid URL'))
result3 = await fetch_specific("ok") # -> Ok('Data from ok')
result4 = await fetch_specific("timeout") # -> Err(ConnectionError('Timeout'))
# result5 = await fetch_specific(123) # -> Raises TypeError (not caught)
# Handle the result
match result1:
case Ok(v):
print(f"Fetched data: {v}")
case Err(ValueError() as e):
print(f"Invalid URL error: {e}")
case Err(e):
print(f"Some other error occurred with fetch_data: {e}")Ok and Err provide methods for transforming and accessing the contained values.
unwrap(): Returns the value if Ok, otherwise raises the contained error. Use cautiously, often within functions already decorated with @safe variants for automatic error propagation.
from safe_result import Err, Ok, Result, safe
ok_res = Ok(42)
err_res = Err(ValueError("Bad data"))
print(ok_res.unwrap()) # -> 42
# err_res.unwrap() # -> Raises ValueError: Bad data
@safe
def combined_op(res1: Result[int, Exception], res2: Result[int, Exception]) -> int:
# unwrap() propagates errors automatically within @safe context
val1 = res1.unwrap()
val2 = res2.unwrap()
return val1 + val2
print(combined_op(Ok(10), Ok(5))) # -> Ok(15)
print(combined_op(Ok(10), Err(ValueError("Fail")))) # -> Err(ValueError('Fail'))unwrap_or(default): Returns the value if Ok, otherwise returns the default value.
print(Ok(42).unwrap_or(0)) # -> 42
print(Err("Error").unwrap_or(0)) # -> 0map(func): Applies func to the value if Ok, returns a new Ok with the result. If Err, returns the original Err unchanged.
print(Ok(5).map(lambda x: x * 2)) # -> Ok(10)
print(Err("Fail").map(lambda x: x * 2)) # -> Err('Fail')map_async(async_func): Applies async_func if Ok. Returns await Ok(await async_func(value)). If Err, returns the original Err.
async def double_async(x):
await asyncio.sleep(0)
return x * 2
async def run_map_async():
print(await Ok(5).map_async(double_async)) # -> Ok(10)
print(await Err("Fail").map_async(double_async)) # -> Err('Fail')and_then(func): Calls func with the value if Ok. func must return a Result. Useful for chaining operations that can fail. If Err, returns the original Err.
def check_positive(n): return Ok(n) if n > 0 else Err("Not positive")
print(Ok(5).and_then(check_positive)) # -> Ok(5)
print(Ok(-1).and_then(check_positive)) # -> Err('Not positive')
print(Err("Fail").and_then(check_positive)) # -> Err('Fail')and_then_async(async_func): Calls async_func with the value if Ok. async_func must return an Awaitable[Result]. If Err, returns the original Err.
async def check_positive_async(n):
await asyncio.sleep(0)
return Ok(n) if n > 0 else Err("Not positive async")
async def run_and_then_async():
print(await Ok(5).and_then_async(check_positive_async)) # -> Ok(5)
print(await Ok(-1).and_then_async(check_positive_async)) # -> Err('Not positive async')
print(await Err("Fail").and_then_async(check_positive_async)) # -> Err('Fail')flatten(): Converts Result[Result[T, E], E] to Result[T, E]. Flattens nested Ok(Ok(value)) to Ok(value) and Ok(Err(error)) to Err(error). Has no effect on non-nested Result or Err.
print(Ok(Ok(42)).flatten()) # -> Ok(42)
print(Ok(Err("Inner")).flatten()) # -> Err('Inner')
print(Err("Outer").flatten()) # -> Err('Outer')
print(Ok(10).flatten()) # -> Ok(10)err_type(result, ExceptionType): Type guard that checks if a Result is an Err containing a specific exception type (or subtype).
from safe_result import err_type
result = Err(ValueError("Invalid input"))
if err_type(result, ValueError):
print("It's a ValueError!") # -> True
if err_type(result, TypeError):
print("It's a TypeError!") # -> False (doesn't print)
if err_type(result, Exception):
print("It's an Exception!") # -> Truetraceback_of(result): Returns the formatted traceback string if the Result is an Err containing an Exception, otherwise returns an empty string.
from safe_result import safe, traceback_of
@safe
def cause_error():
return 1 / 0
error_result = cause_error() # -> Err(ZeroDivisionError('division by zero'))
if not ok(error_result):
tb = traceback_of(error_result)
print(f"Error occurred:\n{tb}")
# Prints the full traceback leading to the ZeroDivisionErrorHere's a practical example using httpx for HTTP requests with proper error handling:
import asyncio
import httpx
from safe_result import safe_async_with, Ok, Err, err_type, traceback_of
# Only catch specific network/HTTP errors
@safe_async_with(httpx.TimeoutException, httpx.HTTPStatusError, ConnectionError)
async def fetch_api_data(url: str, timeout: float = 5.0) -> dict:
async with httpx.AsyncClient() as client:
response = await client.get(url, timeout=timeout)
response.raise_for_status() # Raises HTTPStatusError for 4XX/5XX responses
return response.json()
async def main():
# Example with timeout
result_timeout = await fetch_api_data("https://httpbin.org/delay/10", timeout=2.0)
match result_timeout:
case Ok(data):
print(f"Data received: {data}")
case Err(httpx.TimeoutException):
print("Request timed out - the server took too long to respond")
case Err(httpx.HTTPStatusError as e):
print(f"HTTP Error: {e.response.status_code} for URL: {e.request.url}")
case Err(e): # Catch other specified errors like ConnectionError
print(f"Network error: {e}")
print(traceback_of(result_timeout)) # Print traceback for unexpected errors
# Example with success
result_ok = await fetch_api_data("https://httpbin.org/json")
if ok(result_ok):
print(f"Successfully fetched JSON data: {result_ok.value.get('slideshow', {}).get('title')}")MIT