python-result-type
Advanced tools
A functional programming Result type for Python with Success/Failure variants and async support
A functional programming Result type for Python, inspired by Rust's Result<T, E> and similar to PyMonad's Either, but with more intuitive naming (Success/Failure instead of Right/Left).
Success and Failure instead of cryptic Right/LeftOk and Err aliases for Rust developersResult[T, E].then() method or >> operator for clean chainingAsyncResult wrapperpip install python-result-type
from result_type import Success, Failure, Result
def divide(a: float, b: float) -> Result[float, str]:
if b == 0:
return Failure("Division by zero")
return Success(a / b)
# Success case
result = divide(10, 2)
if result.is_success():
print(f"Result: {result.value}") # Result: 5.0
else:
print(f"Error: {result.error}")
# Failure case
result = divide(10, 0)
if result.is_failure():
print(f"Error: {result.error}") # Error: Division by zero
Chain operations that can fail using .then() or the >> operator:
from result_type import Success, Failure
def divide(a: float, b: float) -> Result[float, str]:
if b == 0:
return Failure("Division by zero")
return Success(a / b)
def multiply_by_2(x: float) -> Result[float, str]:
return Success(x * 2)
def subtract_1(x: float) -> Result[float, str]:
if x < 1:
return Failure("Result would be negative")
return Success(x - 1)
# Method 1: Using .then() method
result = (
divide(10, 2)
.then(multiply_by_2)
.then(subtract_1)
.map(lambda x: x + 5)
)
# Method 2: Using >> operator (cleaner syntax)
result = divide(10, 2) >> multiply_by_2 >> subtract_1
# Method 3: Mixed approach
result = (
divide(10, 2)
>> multiply_by_2
.map(lambda x: x + 10) # Transform without failure
>> subtract_1
)
if result.is_success():
print(f"Final result: {result.value}")
else:
print(f"Error occurred: {result.error}")
Automatically handle exceptions with safe_call:
from result_type import safe_call
# Wrap risky function calls
result = safe_call(
lambda: 10 / 0,
"Math operation failed"
)
if result.is_failure():
print(result.error) # "Math operation failed: division by zero"
# Use as decorator
from result_type import safe_call_decorator
@safe_call_decorator("Database error")
def risky_database_operation():
# Some operation that might throw
return fetch_user_from_db()
result = risky_database_operation()
For developers familiar with Rust's Result<T, E>, this library provides Ok and Err aliases for Success and Failure:
from result_type import Ok, Err, ok, err
def divide_rust_style(a: float, b: float):
if b == 0:
return Err("Division by zero")
return Ok(a / b)
# Usage is identical to Success/Failure
result = divide_rust_style(10, 2)
if result.is_success():
print(f"Result: {result.value}") # Result: 5.0
# Chain operations using Rust-style
def multiply_by_2(x):
return Ok(x * 2)
def validate_positive(x):
if x > 0:
return Ok(x)
return Err("Must be positive")
result = Ok(5) >> multiply_by_2 >> validate_positive
if result.is_success():
print(f"Final: {result.value}") # Final: 10
# Helper functions with Rust naming
success_result = ok(42) # Same as Success(42)
error_result = err("oops") # Same as Failure("oops")
# Mix and match - they're the same types!
mixed_result = Ok(10) >> (lambda x: Success(x * 2)) # Works perfectly
print(Ok(42) == Success(42)) # True
Result[T, E]Abstract base class representing either success or failure.
Methods:
is_success() -> bool - Check if result is Successis_failure() -> bool - Check if result is Failurethen(func: Callable[[T], Result[U, E]]) -> Result[U, E] - Chain operationsmap(func: Callable[[T], U]) -> Result[U, E] - Transform success valuemap_error(func: Callable[[E], F]) -> Result[T, F] - Transform error valueunwrap() -> T - Extract value or raise exceptionunwrap_or(default: T) -> T - Extract value or return defaultunwrap_or_else(func: Callable[[E], T]) -> T - Extract value or compute from errorSuccess[T]Represents successful result containing a value.
success_result = Success(42)
print(success_result.value) # 42
print(success_result.is_success()) # True
Failure[E]Represents failed result containing an error.
failure_result = Failure("Something went wrong")
print(failure_result.error) # "Something went wrong"
print(failure_result.is_failure()) # True
Ok and ErrType aliases for Success and Failure respectively.
from result_type import Ok, Err
# These are identical to Success/Failure
ok_result = Ok(42) # Same as Success(42)
err_result = Err("error") # Same as Failure("error")
ok(value: T) -> Success[T] and err(error: E) -> Failure[E]Helper functions with Rust-style naming.
from result_type import ok, err
result = ok(42) # Same as success(42) or Success(42)
error = err("oops") # Same as failure("oops") or Failure("oops")
success(value: T) -> Success[T]Create a Success result.
from result_type import success
result = success(42) # Same as Success(42)
failure(error: E) -> Failure[E]Create a Failure result.
from result_type import failure
result = failure("error") # Same as Failure("error")
safe_call(func: Callable[[], T], error_msg: str = None) -> Result[T, str]Safely call a function that might raise exceptions.
from result_type import safe_call
result = safe_call(lambda: risky_operation())
if result.is_failure():
print(f"Operation failed: {result.error}")
safe_call_decorator(error_msg: str = None)Decorator version of safe_call.
from result_type import safe_call_decorator
@safe_call_decorator("API call failed")
def call_external_api():
return requests.get("https://api.example.com").json()
result = call_external_api() # Returns Result[dict, str]
When chaining operations, errors automatically propagate:
result = (
Success(10)
>> (lambda x: Failure("Something went wrong")) # This fails
>> (lambda x: Success(x * 2)) # This won't execute
>> (lambda x: Success(x + 1)) # Neither will this
)
print(result.error) # "Something went wrong"
Exceptions in chained operations are automatically converted to Failure:
def risky_operation(x: int) -> Result[int, str]:
return Success(x / 0) # This will raise ZeroDivisionError
result = Success(10) >> risky_operation
print(result.is_failure()) # True
print(type(result.error)) # <class 'ZeroDivisionError'>
The library includes full async/await support for modern Python applications with the AsyncResult wrapper:
import asyncio
from result_type import Success, Failure, Result
from result_type.async_result import AsyncResult, async_safe_call
async def fetch_user(user_id: int) -> Result[dict, str]:
await asyncio.sleep(0.1) # Simulate API call
if user_id <= 0:
return Failure("Invalid user ID")
return Success({"id": user_id, "name": f"User {user_id}"})
# Use async operations
result = await fetch_user(123)
if result.is_success():
print(f"Found user: {result.value}")
Chain async and sync operations seamlessly:
async def fetch_user_posts(user: dict) -> Result[list, str]:
await asyncio.sleep(0.1)
return Success([f"Post {i}" for i in range(3)])
def format_summary(posts: list) -> Result[str, str]:
return Success(f"User has {len(posts)} posts")
# Chain async and sync operations
pipeline = (AsyncResult(fetch_user(123))
.then_async(fetch_user_posts) # Async operation
.then_sync(format_summary)) # Sync operation
result = await pipeline.resolve()
if result.is_success():
print(result.value) # "User has 3 posts"
Handle async exceptions safely:
from result_type.async_result import async_safe_call, async_safe_call_decorator
# Function approach
async def risky_api_call():
# Might raise an exception
return await some_external_api()
result = await async_safe_call(risky_api_call, "API Error")
# Decorator approach
@async_safe_call_decorator("Database Error")
async def database_operation():
return await db.fetch_data()
result = await database_operation() # Returns Result[Any, str]
Process multiple async operations concurrently:
from result_type.async_result import gather_results
async def fetch_data(source: str) -> Result[str, str]:
await asyncio.sleep(0.1)
return Success(f"Data from {source}")
# Gather results - stops at first failure
async_operations = [
AsyncResult(fetch_data("source1")),
AsyncResult(fetch_data("source2")),
AsyncResult(fetch_data("source3")),
]
combined = await gather_results(*async_operations)
if combined.is_success():
print(combined.value) # ["Data from source1", "Data from source2", "Data from source3"]
Convert any awaitable to an AsyncResult:
from result_type.async_result import from_awaitable
async def regular_async_function():
return {"data": "success"}
# Convert to AsyncResult with error handling
async_result = await from_awaitable(regular_async_function(), "Operation failed")
result = await async_result.resolve()
# PyMonad Either (less intuitive)
from pymonad.either import Left, Right
result = Right(42) # Success
result = Left("error") # Failure
# This library (more readable)
from result_type import Success, Failure
result = Success(42) # Clear success intent
result = Failure("error") # Clear failure intent
# Traditional exception handling
try:
result = risky_operation()
result = transform(result)
result = another_transform(result)
except Exception as e:
handle_error(e)
# With Result type
result = (
safe_call(risky_operation)
>> safe_transform
>> safe_another_transform
)
if result.is_failure():
handle_error(result.error)
from result_type import Result, Success, Failure
def fetch_user(user_id: str) -> Result[dict, str]:
try:
user = database.users.find_one({"_id": user_id})
if not user:
return Failure("User not found")
return Success(user)
except Exception as e:
return Failure(f"Database error: {e}")
def validate_user(user: dict) -> Result[dict, str]:
if not user.get("is_active"):
return Failure("User is inactive")
return Success(user)
def get_user_permissions(user: dict) -> Result[list, str]:
permissions = user.get("permissions", [])
if not permissions:
return Failure("User has no permissions")
return Success(permissions)
# Chain the operations
result = (
fetch_user("user123")
>> validate_user
>> get_user_permissions
)
if result.is_success():
print(f"User permissions: {result.value}")
else:
print(f"Failed to get permissions: {result.error}")
import requests
from result_type import safe_call, Result, Success, Failure
def fetch_weather(city: str) -> Result[dict, str]:
return safe_call(
lambda: requests.get(f"http://api.weather.com/{city}").json(),
f"Failed to fetch weather for {city}"
)
def extract_temperature(weather_data: dict) -> Result[float, str]:
try:
temp = weather_data["current"]["temperature"]
return Success(float(temp))
except (KeyError, ValueError, TypeError) as e:
return Failure(f"Invalid weather data: {e}")
def celsius_to_fahrenheit(celsius: float) -> Result[float, str]:
return Success(celsius * 9/5 + 32)
# Chain API call and transformations
result = (
fetch_weather("London")
>> extract_temperature
>> celsius_to_fahrenheit
)
if result.is_success():
print(f"Temperature in Fahrenheit: {result.value}")
else:
print(f"Error: {result.error}")
from pathlib import Path
from result_type import safe_call, Result
def read_config_file(path: str) -> Result[dict, str]:
def _read_and_parse():
content = Path(path).read_text()
return json.loads(content)
return safe_call(_read_and_parse, f"Failed to read config from {path}")
def validate_config(config: dict) -> Result[dict, str]:
required_fields = ["api_key", "database_url", "port"]
missing = [field for field in required_fields if field not in config]
if missing:
return Failure(f"Missing required fields: {missing}")
return Success(config)
def start_application(config: dict) -> Result[str, str]:
# Application startup logic here
return Success(f"Application started on port {config['port']}")
# Chain configuration loading and validation
result = (
read_config_file("config.json")
>> validate_config
>> start_application
)
if result.is_success():
print(result.value) # "Application started on port 8080"
else:
print(f"Startup failed: {result.error}")
# Install development dependencies
pip install python-result-type[dev]
# Run tests
pytest
# Run tests with coverage
pytest --cov=result_type --cov-report=html
# Run type checking
mypy result_type
# Format code
black result_type tests
This library is fully typed and compatible with mypy:
from result_type import Result
def typed_operation(x: int) -> Result[str, str]:
if x < 0:
return Failure("Negative numbers not allowed")
return Success(str(x))
# mypy will catch type errors
result: Result[str, str] = typed_operation(42)
Contributions are welcome! Please read our Contributing Guide for details.
This project is licensed under the MIT License - see the LICENSE file for details.
"Result[T, E]" has no attribute "error" when accessing result.error"Result[T, E]" has no attribute "value" when accessing result.valuevalue and error properties to the abstract Result class for better type checkingOk and Err for Success and Failureok() and err() with Rust-style namingAsyncResult wrapperasync_safe_call and async_safe_call_decorator for async exception handling.then_async() and .then_sync()gather_results() for concurrent async operationsfrom_awaitable() to convert regular awaitables to AsyncResult.then() and >> operatorFAQs
A functional programming Result type for Python with Success/Failure variants and async support
We found that python-result-type demonstrated a healthy version release cadence and project activity because the last version was released less than a year ago. It has 1 open source maintainer collaborating on the project.
Did you know?
Socket for GitHub automatically highlights issues in each pull request and monitors the health of all your open source dependencies. Discover the contents of your packages and block harmful activity before you install or update your dependencies.
Security News
/Product
Socket Firewall Free is now bundled into Docker Hardened Images, adding build-time and dependency-install supply chain protection on top of hardened base images for Node.js, Python, and Rust.
Security News
Season’s greetings from Socket, and here’s to a calm end of year: clean dependencies, boring pipelines, no surprises.
Research
/Security News
Impostor NuGet package Tracer.Fody.NLog typosquats Tracer.Fody and its author, using homoglyph tricks, and exfiltrates Stratis wallet JSON/passwords to a Russian IP address.