# resilient-result
[](https://badge.fury.io/py/resilient-result)
[](https://www.python.org/downloads/)
[](https://opensource.org/licenses/MIT)
[](#testing)
**Beautiful resilient decorators that return Result types instead of throwing exceptions.**
```python
from resilient_result import resilient, Ok, Err, unwrap
@resilient(retries=3, timeout=5)
async def call_api(url: str) -> str:
return await http.get(url) # Exceptions become Result[str, Exception]
result = await call_api("https://api.example.com")
if result.success:
print(result.data) # The API response
else:
print(f"Failed: {result.error}") # The exception that occurred
```
**What just happened?** Function ran up to 3 times with exponential backoff, timed out after 5s, and returned `Result[str, Exception]` instead of throwing.
## Installation
```bash
pip install resilient-result
```
## Core Patterns
### Basic Resilience
```python
# Simple retry with defaults
@resilient()
async def might_fail():
if random.random() < 0.3:
raise Exception("Oops!")
return "success"
# Returns Ok("success") or Err(Exception("Oops!"))
result = await might_fail()
```
### Built-in Patterns
```python
# Network calls with smart retry logic
@resilient.network(retries=2)
async def fetch_data(url: str):
return await httpx.get(url)
# JSON parsing with error recovery
@resilient.parsing()
async def parse_response(text: str):
return json.loads(text)
# Circuit breaker protection
@resilient.circuit(failures=5, window=60)
async def external_service():
return await service.call()
# Rate limiting with token bucket
@resilient.rate_limit(rps=10.0, burst=5)
async def api_call():
return await external_api()
```
## Result Type System
### Type-Safe Error Handling
```python
from resilient_result import Result, Ok, Err
# Functions return Result[T, E] instead of throwing
def divide(a: int, b: int) -> Result[int, str]:
if b == 0:
return Err("Division by zero")
return Ok(a // b)
# Pattern matching for elegant handling
result = divide(10, 2)
match result:
case Ok(value):
print(f"Result: {value}")
case Err(error):
print(f"Error: {error}")
```
### Smart Result Detection
```python
# Already returns Result? Passes through unchanged
@resilient(retries=2)
async def already_result() -> Result[str, ValueError]:
return Ok("data") # Unchanged: Ok("data")
# Regular return? Auto-wrapped in Ok()
@resilient(retries=2)
async def regular_return() -> str:
return "data" # Becomes: Ok("data")
# Exception raised? Becomes Err()
@resilient(retries=2)
async def might_throw() -> str:
raise ValueError("oops") # Becomes: Err(ValueError("oops"))
```
## Extensibility - Registry System
### Creating Custom Patterns
```python
from resilient_result import resilient, decorator
# Define domain-specific handler
async def llm_handler(error):
error_str = str(error).lower()
if "rate_limit" in error_str:
await asyncio.sleep(60) # Wait for rate limit reset
return None # Trigger retry
if "context_length" in error_str:
return False # Don't retry context errors
return None # Retry other errors
# Create pattern factory
def llm_pattern(retries=3, **kwargs):
return decorator(handler=llm_handler, retries=retries, **kwargs)
# Register with resilient-result
resilient.register("llm", llm_pattern)
# Beautiful usage
@resilient.llm(retries=5, timeout=30)
async def call_openai(prompt: str):
return await openai.create(prompt=prompt)
```
### Real-World Extension: AI Agent Patterns
```python
# Cogency extends resilient-result for AI-specific resilience
from cogency.resilience import safe # Built on resilient-result
@safe.reasoning(retries=2) # Fallback: deep → fast mode
async def llm_reasoning(state):
return await llm.generate(state.prompt)
@safe.memory() # Graceful memory degradation
async def store_context(data):
return await vector_db.store(data)
# Both @safe.reasoning() and @resilient.reasoning() work identically
# Proving the extensibility architecture works beautifully
```
## Performance & Architecture
### Performance Characteristics
- **Overhead**: ~0.1ms per decorated call
- **Memory**: ~200 bytes per Result object
- **Concurrency**: Thread-safe, async-first design
- **Test suite**: Comprehensive coverage, <2s runtime
### v0.2.2 Status: Ergonomic & Powerful
✅ **Proven extensible architecture** - Registry system enables domain-specific patterns
✅ **Beautiful decorator API** - Clean `@resilient.pattern()` and `@resilient` syntax
✅ **Type-safe Result system** - Ok/Err prevents ignored errors
✅ **Real-world proven** - Successfully integrated with production AI systems
✅ **Enhanced ergonomics** - `unwrap()`, `Result.collect()`, and fallback patterns
## New in v0.2.2: Enhanced Developer Experience
### `unwrap()` Function - Clean Result Extraction
```python
from resilient_result import unwrap
@resilient
async def api_call():
return "success"
# Clean extraction - raises exception if failed
data = unwrap(await api_call()) # "success"
```
### `Result.collect()` - Parallel Operations Made Easy
```python
# Run multiple async operations, collect all results
operations = [
fetch_user(user_id),
fetch_profile(user_id),
fetch_settings(user_id)
]
result = await Result.collect(operations)
if result.success:
user, profile, settings = result.data # All succeeded
else:
print(f"Operation failed: {result.error}") # First failure
```
### `@resilient.fallback()` - Mode Switching Pattern
```python
# Automatically fallback between modes on error
@resilient.fallback("complexity", "simple", retries=2)
async def adaptive_processing(state):
if state.complexity == "advanced":
return await complex_algorithm(state)
else:
return await simple_algorithm(state)
# On error, automatically switches state.complexity to "simple" and retries
```
### Enhanced `@resilient` Syntax
```python
# Both syntaxes work identically
@resilient # Clean, no parentheses
@resilient() # Traditional with parentheses
@resilient(retries=5) # With parameters
```
Current patterns provide solid foundation with basic implementations suitable for development and basic production use.
*Production-grade pattern enhancements planned for v0.3.0 - see [roadmap](docs/roadmap.md)*
## When to Use
✅ **Perfect for**:
- API clients and external service calls
- Data processing pipelines
- AI/LLM applications with retry logic
- Microservices with resilience requirements
- Any async operations that might fail
❌ **Not ideal for**:
- High-frequency inner loops (0.1ms overhead)
- Simple scripts (adds complexity)
- Teams preferring exception-based patterns
## Advanced Examples
### Composing Multiple Patterns
```python
# Stack decorators for layered resilience
@resilient.rate_limit(rps=5)
@resilient.circuit(failures=3)
@resilient.network(retries=2)
async def robust_api_call(endpoint: str):
return await http.get(f"https://api.service.com/{endpoint}")
```
### Custom Error Types
```python
class APIError(Exception):
pass
@resilient(retries=3, error_type=APIError, timeout=10)
async def typed_api_call(data: dict):
response = await http.post("/api/endpoint", json=data)
return response.json()
# Returns Result[dict, APIError] - type-safe!
```
### Sync Function Support
```python
@resilient(retries=3)
def sync_operation(data: str) -> str:
if random.random() < 0.3:
raise Exception("Sync failure")
return f"processed: {data}"
# Also returns Result[str, Exception]
result = sync_operation("test")
```
## Testing Made Easy
```python
# No more exception mocking - just check Result values
async def test_api_call():
result = await call_api("https://fake-url")
assert isinstance(result, Result)
if result.success:
assert "data" in result.data
else:
assert "network" in str(result.error).lower()
```
## License
MIT - Build amazing resilient systems! 🚀
Raw data
{
"_id": null,
"home_page": "https://github.com/iteebz/resilient-result",
"name": "resilient-result",
"maintainer": null,
"docs_url": null,
"requires_python": "<4.0,>=3.8",
"maintainer_email": null,
"keywords": "result, error-handling, resilient, retry, timeout, async",
"author": "resilient-result contributors",
"author_email": null,
"download_url": "https://files.pythonhosted.org/packages/3f/36/f9abb15cd7813bef724cf1659f796fbe3319a6909806df82f2e098dcdce5/resilient_result-0.2.2.tar.gz",
"platform": null,
"description": "# resilient-result\n\n[](https://badge.fury.io/py/resilient-result)\n[](https://www.python.org/downloads/)\n[](https://opensource.org/licenses/MIT)\n[](#testing)\n\n**Beautiful resilient decorators that return Result types instead of throwing exceptions.**\n\n```python\nfrom resilient_result import resilient, Ok, Err, unwrap\n\n@resilient(retries=3, timeout=5)\nasync def call_api(url: str) -> str:\n return await http.get(url) # Exceptions become Result[str, Exception]\n\nresult = await call_api(\"https://api.example.com\")\nif result.success:\n print(result.data) # The API response\nelse:\n print(f\"Failed: {result.error}\") # The exception that occurred\n```\n\n**What just happened?** Function ran up to 3 times with exponential backoff, timed out after 5s, and returned `Result[str, Exception]` instead of throwing.\n\n## Installation\n\n```bash\npip install resilient-result\n```\n\n## Core Patterns\n\n### Basic Resilience\n```python\n# Simple retry with defaults\n@resilient()\nasync def might_fail():\n if random.random() < 0.3:\n raise Exception(\"Oops!\")\n return \"success\"\n\n# Returns Ok(\"success\") or Err(Exception(\"Oops!\"))\nresult = await might_fail()\n```\n\n### Built-in Patterns\n```python\n# Network calls with smart retry logic\n@resilient.network(retries=2)\nasync def fetch_data(url: str):\n return await httpx.get(url)\n\n# JSON parsing with error recovery\n@resilient.parsing()\nasync def parse_response(text: str):\n return json.loads(text)\n\n# Circuit breaker protection\n@resilient.circuit(failures=5, window=60)\nasync def external_service():\n return await service.call()\n\n# Rate limiting with token bucket\n@resilient.rate_limit(rps=10.0, burst=5)\nasync def api_call():\n return await external_api()\n```\n\n## Result Type System\n\n### Type-Safe Error Handling\n```python\nfrom resilient_result import Result, Ok, Err\n\n# Functions return Result[T, E] instead of throwing\ndef divide(a: int, b: int) -> Result[int, str]:\n if b == 0:\n return Err(\"Division by zero\")\n return Ok(a // b)\n\n# Pattern matching for elegant handling\nresult = divide(10, 2)\nmatch result:\n case Ok(value):\n print(f\"Result: {value}\")\n case Err(error):\n print(f\"Error: {error}\")\n```\n\n### Smart Result Detection\n```python\n# Already returns Result? Passes through unchanged\n@resilient(retries=2)\nasync def already_result() -> Result[str, ValueError]:\n return Ok(\"data\") # Unchanged: Ok(\"data\")\n\n# Regular return? Auto-wrapped in Ok()\n@resilient(retries=2) \nasync def regular_return() -> str:\n return \"data\" # Becomes: Ok(\"data\")\n\n# Exception raised? Becomes Err()\n@resilient(retries=2)\nasync def might_throw() -> str:\n raise ValueError(\"oops\") # Becomes: Err(ValueError(\"oops\"))\n```\n\n## Extensibility - Registry System\n\n### Creating Custom Patterns\n```python\nfrom resilient_result import resilient, decorator\n\n# Define domain-specific handler\nasync def llm_handler(error):\n error_str = str(error).lower()\n if \"rate_limit\" in error_str:\n await asyncio.sleep(60) # Wait for rate limit reset\n return None # Trigger retry\n if \"context_length\" in error_str:\n return False # Don't retry context errors\n return None # Retry other errors\n\n# Create pattern factory\ndef llm_pattern(retries=3, **kwargs):\n return decorator(handler=llm_handler, retries=retries, **kwargs)\n\n# Register with resilient-result\nresilient.register(\"llm\", llm_pattern)\n\n# Beautiful usage\n@resilient.llm(retries=5, timeout=30)\nasync def call_openai(prompt: str):\n return await openai.create(prompt=prompt)\n```\n\n### Real-World Extension: AI Agent Patterns\n```python\n# Cogency extends resilient-result for AI-specific resilience\nfrom cogency.resilience import safe # Built on resilient-result\n\n@safe.reasoning(retries=2) # Fallback: deep \u2192 fast mode\nasync def llm_reasoning(state):\n return await llm.generate(state.prompt)\n\n@safe.memory() # Graceful memory degradation\nasync def store_context(data): \n return await vector_db.store(data)\n\n# Both @safe.reasoning() and @resilient.reasoning() work identically\n# Proving the extensibility architecture works beautifully\n```\n\n## Performance & Architecture\n\n### Performance Characteristics\n- **Overhead**: ~0.1ms per decorated call\n- **Memory**: ~200 bytes per Result object \n- **Concurrency**: Thread-safe, async-first design\n- **Test suite**: Comprehensive coverage, <2s runtime\n\n### v0.2.2 Status: Ergonomic & Powerful\n\u2705 **Proven extensible architecture** - Registry system enables domain-specific patterns \n\u2705 **Beautiful decorator API** - Clean `@resilient.pattern()` and `@resilient` syntax \n\u2705 **Type-safe Result system** - Ok/Err prevents ignored errors \n\u2705 **Real-world proven** - Successfully integrated with production AI systems\n\u2705 **Enhanced ergonomics** - `unwrap()`, `Result.collect()`, and fallback patterns\n\n## New in v0.2.2: Enhanced Developer Experience\n\n### `unwrap()` Function - Clean Result Extraction\n```python\nfrom resilient_result import unwrap\n\n@resilient\nasync def api_call():\n return \"success\"\n\n# Clean extraction - raises exception if failed\ndata = unwrap(await api_call()) # \"success\"\n```\n\n### `Result.collect()` - Parallel Operations Made Easy\n```python\n# Run multiple async operations, collect all results\noperations = [\n fetch_user(user_id),\n fetch_profile(user_id),\n fetch_settings(user_id)\n]\n\nresult = await Result.collect(operations)\nif result.success:\n user, profile, settings = result.data # All succeeded\nelse:\n print(f\"Operation failed: {result.error}\") # First failure\n```\n\n### `@resilient.fallback()` - Mode Switching Pattern \n```python\n# Automatically fallback between modes on error\n@resilient.fallback(\"complexity\", \"simple\", retries=2)\nasync def adaptive_processing(state):\n if state.complexity == \"advanced\":\n return await complex_algorithm(state)\n else:\n return await simple_algorithm(state)\n\n# On error, automatically switches state.complexity to \"simple\" and retries\n```\n\n### Enhanced `@resilient` Syntax\n```python\n# Both syntaxes work identically\n@resilient # Clean, no parentheses \n@resilient() # Traditional with parentheses\n@resilient(retries=5) # With parameters\n```\n\nCurrent patterns provide solid foundation with basic implementations suitable for development and basic production use.\n\n*Production-grade pattern enhancements planned for v0.3.0 - see [roadmap](docs/roadmap.md)*\n\n## When to Use\n\n\u2705 **Perfect for**:\n- API clients and external service calls\n- Data processing pipelines \n- AI/LLM applications with retry logic\n- Microservices with resilience requirements\n- Any async operations that might fail\n\n\u274c **Not ideal for**:\n- High-frequency inner loops (0.1ms overhead)\n- Simple scripts (adds complexity)\n- Teams preferring exception-based patterns\n\n## Advanced Examples\n\n### Composing Multiple Patterns\n```python\n# Stack decorators for layered resilience\n@resilient.rate_limit(rps=5)\n@resilient.circuit(failures=3)\n@resilient.network(retries=2)\nasync def robust_api_call(endpoint: str):\n return await http.get(f\"https://api.service.com/{endpoint}\")\n```\n\n### Custom Error Types\n```python\nclass APIError(Exception):\n pass\n\n@resilient(retries=3, error_type=APIError, timeout=10)\nasync def typed_api_call(data: dict):\n response = await http.post(\"/api/endpoint\", json=data)\n return response.json()\n\n# Returns Result[dict, APIError] - type-safe!\n```\n\n### Sync Function Support\n```python\n@resilient(retries=3)\ndef sync_operation(data: str) -> str:\n if random.random() < 0.3:\n raise Exception(\"Sync failure\")\n return f\"processed: {data}\"\n\n# Also returns Result[str, Exception]\nresult = sync_operation(\"test\")\n```\n\n## Testing Made Easy\n\n```python\n# No more exception mocking - just check Result values\nasync def test_api_call():\n result = await call_api(\"https://fake-url\")\n \n assert isinstance(result, Result)\n if result.success:\n assert \"data\" in result.data\n else:\n assert \"network\" in str(result.error).lower()\n```\n\n## License\n\nMIT - Build amazing resilient systems! \ud83d\ude80",
"bugtrack_url": null,
"license": "MIT",
"summary": "Resilient decorators that return Result types instead of throwing exceptions",
"version": "0.2.2",
"project_urls": {
"Homepage": "https://github.com/iteebz/resilient-result",
"Repository": "https://github.com/iteebz/resilient-result"
},
"split_keywords": [
"result",
" error-handling",
" resilient",
" retry",
" timeout",
" async"
],
"urls": [
{
"comment_text": "",
"digests": {
"blake2b_256": "b8e14e13e7a2b9013c87c856aa65d1ab8853dcf7d01f5d706db940494fcc56cb",
"md5": "2a0d915993d43e84d650eb1c073b8c2f",
"sha256": "02f8a197024b2da59fe414d60eb731d26c8afb03ec7a74f1a3dfb6c1b3ca1829"
},
"downloads": -1,
"filename": "resilient_result-0.2.2-py3-none-any.whl",
"has_sig": false,
"md5_digest": "2a0d915993d43e84d650eb1c073b8c2f",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": "<4.0,>=3.8",
"size": 12154,
"upload_time": "2025-07-27T02:45:07",
"upload_time_iso_8601": "2025-07-27T02:45:07.328977Z",
"url": "https://files.pythonhosted.org/packages/b8/e1/4e13e7a2b9013c87c856aa65d1ab8853dcf7d01f5d706db940494fcc56cb/resilient_result-0.2.2-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": "",
"digests": {
"blake2b_256": "3f36f9abb15cd7813bef724cf1659f796fbe3319a6909806df82f2e098dcdce5",
"md5": "3fcc6b216ab18f4bbfd0402a7f4149e5",
"sha256": "3c8f6b507a2ab3a572e919cfb4ec216b64bc5e34647587c2bddafcf86a023d09"
},
"downloads": -1,
"filename": "resilient_result-0.2.2.tar.gz",
"has_sig": false,
"md5_digest": "3fcc6b216ab18f4bbfd0402a7f4149e5",
"packagetype": "sdist",
"python_version": "source",
"requires_python": "<4.0,>=3.8",
"size": 12063,
"upload_time": "2025-07-27T02:45:08",
"upload_time_iso_8601": "2025-07-27T02:45:08.285773Z",
"url": "https://files.pythonhosted.org/packages/3f/36/f9abb15cd7813bef724cf1659f796fbe3319a6909806df82f2e098dcdce5/resilient_result-0.2.2.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2025-07-27 02:45:08",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "iteebz",
"github_project": "resilient-result",
"travis_ci": false,
"coveralls": false,
"github_actions": false,
"lcname": "resilient-result"
}
JSON
