feat: add supporting types for error handling

- add a result type similar to the Rust Result type from
https://github.com/rustedpy/result
- add a Report class to propagate error information back on the stack
with return types instead of exceptions

refs #123
This commit is contained in:
fusion44 2025-03-16 17:16:20 +01:00 committed by fusion44
parent 8b372ecb47
commit ef9bfbefe4
18 changed files with 3743 additions and 1 deletions

2
.gitignore vendored
View file

@ -11,7 +11,7 @@ pyrightconfig.json
test_env_data
# Nix
result
./result/
# Devenv
.devenv*
devenv.local.nix

View file

@ -0,0 +1,293 @@
# Python Error Stack
A library that emulates Rust's `error_stack` crate, providing nested error context and rich error reporting.
## Overview
The Error Stack library provides a robust way to handle errors in Python applications, inspired by Rust's `error_stack` crate. It allows for:
1. Creating layered error reports with nested context
2. Adding arbitrary attachments to errors, with support for sensitive data handling
3. Capturing source location information automatically
4. Pretty-printing errors in a hierarchical style that shows the error propagation path
5. Proper formatting of multi-line attachments with continuation lines
## Core Components
### `Attachment`
Represents data attached to an error frame:
- Contains a value, optional name, and sensitive flag
- Can be marked as sensitive to prevent accidental exposure in logs
- Handles multi-line text with proper formatting but doesn't preserve whitespace withing the multi-line text
### `Frame`
Represents a single error or context frame in the error stack. Contains:
- A message
- Optional exception
- Source code location
- Arbitrary attachments (including sensitive data)
### `Report`
The main error report containing a stack of frames. Supports:
- Adding new context frames
- Attaching data to frames
- Pretty printing in a hierarchical format
- Capturing exception tracebacks
- Control over display of sensitive information
## Usage Examples
### Basic Usage
```python
from app.api.error_report.report import Report
try:
# Some operation that might fail
with open("non_existent_file.txt", "r") as f:
content = f.read()
except FileNotFoundError as e:
# Create an error report
err = Report("Could not load configuration", e)
# Print it
print(err)
# Or return it
return err
```
### Adding Nested Context
When errors propagate through different layers of your application, you can add context:
```python
from app.api.error_report.report import Report
def low_level_function():
try:
with open("config.json", "r") as f:
return f.read()
except FileNotFoundError as e:
return Report("Failed to read configuration file", e)
def mid_level_function():
result = low_level_function()
if isinstance(result, Report):
return result.change_context("Configuration loading failed")
# Process the file contents...
return result
def high_level_function():
result = mid_level_function()
if isinstance(result, Report):
return result.change_context("Application initialization error")
# Continue with application...
return result
```
### Using with `Result` Type
For more Rust-like error handling, combine with a Result type:
```python
from app.api.error_report.report import Report
from result import Result, Ok, Err # Use your preferred Result implementation
def read_file(path: str) -> Result[str, Report]:
try:
with open(path, "r") as f:
content = f.read()
return Ok(content)
except FileNotFoundError as e:
return Err(Report(f"Could not read file '{path}'", e))
def process_config() -> Result[dict, Report]:
match read_file("config.json"):
case Ok(content):
# Process content
return Ok({"success": True})
case Err(report):
# Add context and propagate
return Err(report.change_context("Failed to process configuration"))
```
### Attaching Data
You can attach arbitrary data to error reports for additional context:
```python
def process_user_data(user_id: str) -> Result[dict, Report]:
try:
# Process user data
user = db.get_user(user_id)
return Ok(user)
except DatabaseError as e:
err = Report("User data processing failed", e)
# Attach contextual information
err.attach(user_id, "user_id")
err.attach({"attempted_at": datetime.now()}, "metadata")
return Err(err)
```
### Multi-line Attachments
The library properly handles multi-line text in attachments:
```python
err = Report("Failed to process message", error)
err.attach("""User tried to send a message.
This message contained invalid formatting.
Attempted to process anyway but failed.""")
```
This will produce formatted output with continuation lines:
```
Failed to process message
├╴at /path/to/file.py:35:10
├╴User tried to send a message.
│ This message contained invalid formatting.
│ Attempted to process anyway but failed.
╰─▶ [Errno 2] Invalid message format
```
### Handling Sensitive Data
For sensitive information that shouldn't appear in normal logs:
```python
def authenticate_user(username: str, password: str) -> Result[User, Report]:
try:
# Authentication process
user = auth_service.authenticate(username, password)
return Ok(user)
except AuthenticationError as e:
err = Report("Authentication failed", e)
# Safe to include in all logs
err.attach(username, "username")
# Mark sensitive data to be redacted in normal output
# Example purposes, never ever include passwords in logs in production deployments
err.attach(password, "password", sensitive=True)
err.attach({"ip": "192.168.1.1", "api_key": "sk_test_123"}, "connection_info", sensitive=True)
return Err(err)
# Usage:
match authenticate_user("johndoe", "secret123"):
case Ok(user):
print(f"Authenticated: {user.name}")
case Err(report):
# Normal output (safe for logs, no sensitive data)
print(report)
# For debug purposes only, include sensitive data
print(report.format(include_sensitive=True))
# Or with full traceback and sensitive data
print(report.format_verbose())
```
The normal output would show:
```
Authentication failed
├╴at /path/to/file.py:35:10
├╴username: johndoe
├╴password: (Sensitive data omitted)
├╴connection_info: (Sensitive data omitted)
...
```
But the sensitive version would show:
```
Authentication failed
├╴at /path/to/file.py:35:10
├╴username: johndoe
├╴password: secret123
├╴connection_info: {'ip': '192.168.1.1', 'api_key': 'sk_test_123'}
...
```
### Nested Errors Example
Here's a more complete example showing nested error handling through multiple layers:
```python
def level_4_function() -> Result[str, Report]:
try:
with open("missing_file.txt", "r") as f:
content = f.read()
return Ok(content)
except FileNotFoundError as e:
return Err(Report("Something went wrong at level 4", e))
def level_3_function() -> Result[str, Report]:
match level_4_function():
case Ok(content):
return Ok(content)
case Err(report):
return Err(
report.change_context("Changing context at level 3")
.attach("Additional context for debugging")
)
def level_2_function() -> Result[str, Report]:
match level_3_function():
case Ok(content):
return Ok(content)
case Err(report):
return Err(report.change_context("Error occurred at level 2"))
def level_1_function() -> Result[str, Report]:
match level_2_function():
case Ok(content):
return Ok(content)
case Err(report):
return Err(report.change_context("Top level error context"))
# Usage
match level_1_function():
case Ok(content):
print("Success:", content)
case Err(report):
print(report.format()) # Prints the nested error context
```
## Benefits
1. **Clear Error Context**: Creates a hierarchy of error contexts that help track down the root cause.
2. **Rich Debugging Information**: Automatically captures source locations and provides detailed error traces.
3. **Separation of Concerns**: Allows different layers of your application to add appropriate context without losing the original error.
4. **Consistent Error Handling**: Provides a uniform way to handle errors throughout your application.
5. **Security Conscious**: Allows including sensitive data for debugging while preventing accidental exposure in logs.
6. **Readable Formatting**: Maintains proper formatting for multi-line text with continuation lines.
## Potential Downsides
1. **Overhead**: Creating detailed error reports with locations and tracebacks adds computational and memory overhead compared to simple exceptions.
2. **Learning Curve**: The pattern is different from traditional Python exception handling, requiring developers to learn a new approach.
3. **Return Value Checking**: Without using a Result type, you need explicit type checking on return values to determine if you got a Report or a valid result.
4. **Memory Usage**: For long error chains with many attachments, memory usage can grow significantly.
5. **Limited Integration**: Not all third-party libraries support this pattern, requiring adapter code at integration boundaries.
6. **Potential for Data Leakage**: Despite the sensitive data handling, there's a risk of accidentally exposing sensitive information if `include_sensitive=True` is used inappropriately.
7. **Serialization Challenges**: When serializing errors (for logging or API responses), ensuring consistent handling of complex nested data structures requires careful implementation.
## Best Practices
1. Create error reports at the lowest level where exceptions occur
2. Add context as errors propagate up through your application layers
3. Mark sensitive data appropriately using the `sensitive=True` flag
4. Only use `include_sensitive=True` or `format_verbose()` in controlled environments
5. Use multi-line text for detailed explanations that need more than a single line
6. Use with a Result type for more predictable error handling
7. Format reports at the application boundary for logging or user display

View file

@ -0,0 +1,119 @@
from app.external.result_type.src.result import Err, Result, Ok
from app.api.error_report.report import Report
def nested_err_4() -> Result[str, Report]:
try:
with open("this_file_doesnt_exist.txt", "r") as f:
content = f.read()
print(content)
except FileNotFoundError as e:
return Err(Report("Something went wrong at level 4", e))
return Ok("Ok")
def nested_err_3() -> Result[str, Report]:
res = nested_err_4()
match res:
case Ok(_):
print("Level 4 was OK")
case Err(report):
return Err(
report.change_context("Changing context at level 3. Level 4 haz errors")
.attach("This is an attachment with more information at level 3")
.attach(
"""User tried to send a message.
Testing multi line attachments.
More explanation here!!!"""
)
.attach(
{
"data": "this is a sensitive message",
"recipient": "test_recipient",
},
name="input",
sensitive=True,
)
)
return Ok("Ok")
def nested_err_2() -> Result[str, Report]:
res = nested_err_3()
match res:
case Ok(_):
print("Level 3 was OK")
case Err(report):
return Err(
report.change_context("Changing context at level 2. Level 3 haz errors")
.attach("This is an attachment with more information at level 2")
.attach("Even more context")
)
return Ok("Ok")
def nested_err_1() -> Result[str, Report]:
res = nested_err_2()
match res:
case Ok(_):
print("Level 2 was OK")
case Err(report):
return Err(
report.change_context(
"Changing context at level 1. Level 2 or above had errors... teeest"
)
)
return Ok("Ok")
def read_config_file(filename: str) -> Result[bool, Report]:
"""Attempt to read and parse a configuration file."""
try:
with open(filename, "r") as f:
content = f.read()
# Simulate parsing error
if not content.strip():
return Err(Report("Config file not found", ValueError()))
# In real code, you might parse JSON, YAML, etc.
return Ok(True)
except FileNotFoundError as e:
# Wrap the low-level error with context
err = Report(f'could not read file "{filename}"', e)
# You can add additional context or attachments
err.attach(filename, "filename")
print("returning err file not found")
return Err(err)
except ValueError as e:
# Another way to wrap errors
report = Report("Error parsing config", e)
print("parse error")
return Err(report)
def process_configuration() -> Result[bool, Report]:
"""Process application configuration with proper error handling."""
match read_config_file("config.cfg"):
case Ok(value):
print(f"Configuration loaded successfully: {value}")
case Err(report):
report.change_context("Unable to configure the application")
return Err(report)
return Ok(True)
if __name__ == "__main__":
res = nested_err_1()
match res:
case Ok(value):
print("Success")
case Err(report):
print(report.format_verbose())

View file

@ -0,0 +1,295 @@
from typing import Optional, List, Any
import inspect
import traceback
from dataclasses import dataclass, field
@dataclass
class Attachment:
"""Represents an attachment to an error frame."""
value: Any
name: Optional[str] = None
sensitive: bool = True
def __str__(self, include_sensitive: bool = False) -> str:
if self.sensitive and not include_sensitive:
if self.name:
return f"{self.name}: (Sensitive data omitted)"
return "(Sensitive data omitted)"
if self.name:
base = f"{self.name}: {self.value}"
else:
base = f"{self.value}"
return base
@dataclass
class Location:
"""Represents a source code location."""
file: str
line: int
column: int = 0
def __str__(self) -> str:
if self.column:
return f"{self.file}:{self.line}:{self.column}"
return f"{self.file}:{self.line}"
@dataclass
class Frame:
"""Represents a single error frame in the error stack."""
message: str
error: Optional[Exception] = None
location: Optional[Location] = None
attachments: List[Attachment] = field(default_factory=list)
def __post_init__(self):
"""Initialize location if not provided."""
if not self.location:
# Get caller frame information
frame = inspect.currentframe()
for _ in range(3): # Skip our own frames
if frame and frame.f_back:
frame = frame.f_back
else:
frame = None
break
if frame:
self.location = Location(
file=frame.f_code.co_filename,
line=frame.f_lineno,
column=10, # Column is often not available, use a default
)
def attach(
self, value: Any, name: Optional[str] = None, sensitive: bool = True
) -> "Frame":
"""
Attach arbitrary data to this frame.
Args:
value: The data to attach
name: Optional name for the attachment
sensitive: Whether this attachment contains sensitive data
"""
self.attachments.append(Attachment(value, name, sensitive))
return self
def __str__(self) -> str:
return self.message
class Report:
"""Represents an error report containing multiple frames and context."""
def __init__(self, message: str, error: Optional[Exception] = None):
"""Create a new error report with an initial frame."""
self._frames: List[Frame] = []
self._root_error = error
# Create and add the root frame - this is the higher level context/message
self.attach_frame(Frame(message=message))
# If an exception was provided, capture its traceback
self._traceback = None
if error:
self._traceback = (
traceback.extract_tb(error.__traceback__)
if error.__traceback__
else None
)
def attach_frame(self, frame: Frame) -> "Report":
"""Add a new error frame to the report."""
self._frames.append(frame)
return self
def change_context(self, message: str) -> "Report":
"""Add a new context frame to the report."""
return self.attach_frame(Frame(message=message))
def attach(
self, value: Any, name: Optional[str] = None, sensitive: bool = False
) -> "Report":
"""
Attach data to the most recent frame.
Args:
value: The data to attach
name: Optional name for the attachment
sensitive: Whether this attachment contains sensitive data that
should be redacted in normal error reports
"""
if self._frames:
self._frames[-1].attach(value, name, sensitive)
return self
def _format_attachment(
self,
attachment_str: str,
indent: str,
) -> List[str]:
"""
Format an attachment string with proper multi-line handling.
Args:
attachment_str: The attachment string to format
indent: Current indentation level
include_sensitive: Whether to include sensitive data
Returns:
List of formatted lines
"""
# Split the attachment string into lines
lines = attachment_str.split("\n")
result = []
# First line gets the attachment prefix
if lines:
result.append(f"{indent}├╴{lines[0]}")
# Subsequent lines get continuation prefix
for line in lines[1:]:
# Strip leading whitespace from continuation lines for better formatting
cleaned_line = line.lstrip()
if cleaned_line: # Avoid adding empty lines
result.append(f"{indent}{cleaned_line}")
return result
def format(self, include_sensitive: bool = False) -> str:
"""
Format the error report for display with nested contexts.
Args:
include_sensitive: Whether to include sensitive attachment
data in the output
"""
if not self._frames:
return "Empty error report"
# In the expected output, contexts are printed in reverse order (newest first)
frames = list(reversed(self._frames))
# Start with the newest context/frame (the last one added)
top_frame = frames[0]
lines = []
lines.append(str(top_frame))
# Add location for the top frame
if top_frame.location:
lines.append(f"├╴at {top_frame.location}")
# For attachments on the top frame
for attachment in top_frame.attachments:
attachment_lines = self._format_attachment(
attachment.__str__(include_sensitive), ""
)
lines.extend(attachment_lines)
# If we have more than one frame
if len(frames) > 1:
# Start the hierarchical error structure
lines.append(f"╰─▶ {frames[1].message}")
# Process all frames from the second one onwards
current_indent = " "
for i in range(1, len(frames)):
frame = frames[i]
# Add location
if frame.location:
lines.append(f"{current_indent}├╴at {frame.location}")
# Add attachments
for attachment in frame.attachments:
attachment_lines = self._format_attachment(
attachment.__str__(include_sensitive),
current_indent,
)
lines.extend(attachment_lines)
# If not the last frame, add the arrow to the next frame
if i < len(frames) - 1:
lines.append(f"{current_indent}╰─▶ {frames[i + 1].message}")
# If last frame and we have a root error
elif self._root_error:
# Add backtrace info
lines.append(f"{current_indent}")
lines.append(
f"{current_indent}╰╴backtrace "
f"({len(self._traceback) if self._traceback else 0})"
)
# Increase indentation for the next level
current_indent += " "
else:
# If only one frame and we have a root error
if self._root_error:
error_str = str(self.root_error).replace("\n", "\n")
lines.append(f"") # noqa: F541
lines.append(f"╰─▶ {error_str}")
# Add location for error
if top_frame.location and not self._traceback:
lines.append(f" ╰╴at {top_frame.location}")
if top_frame.location and self._traceback:
lines.append(f" ├╴at {top_frame.location}")
# Add backtrace info
if self._traceback:
lines.append(f"") # noqa: F541
lines.append(f" ╰╴backtrace ({len(self._traceback)})")
return "\n".join(lines)
def format_verbose(self, include_sensitive: bool = False) -> str:
"""
Format the error report with full traceback information.
Args:
include_sensitive: Whether to include sensitive attachment data
(defaults to True for verbose mode)
"""
basic_output = self.format(include_sensitive)
if not self._traceback:
return basic_output
# Add the full traceback
tb_lines = []
for frame in self._traceback:
filename, line, func, code = frame
tb_lines.append(f' File "{filename}", line {line}, in {func}')
if code:
tb_lines.append(f" {code}")
return basic_output + "\n" + "\n".join(tb_lines)
def __str__(self) -> str:
"""Default string representation without sensitive data."""
return self.format(include_sensitive=False)
def __repr__(self) -> str:
"""Representation with indication of frames and error presence."""
frame_count = len(self._frames)
has_error = "with error" if self._root_error else "without error"
return f"<Report: {frame_count} frames, {has_error}>"
@property
def frames(self) -> List[Frame]:
"""Access the frames list (useful for testing and debugging)."""
return self._frames.copy()
@property
def root_error(self) -> Optional[Exception]:
"""Access the root error (useful for testing and debugging)."""
return self._root_error

15
app/external/result_type/.gitignore vendored Normal file
View file

@ -0,0 +1,15 @@
.cache/
.coverage
coverage.xml
*.swp
*.pyc
__pycache__
dist/
*.egg-info/
build/
.idea/
.mypy_cache/
venv/
/.tox/
.vscode
pyrightconfig.json

175
app/external/result_type/CHANGELOG.md vendored Normal file
View file

@ -0,0 +1,175 @@
# Changelog
This project follows semantic versioning.
Possible log types:
- `[added]` for new features.
- `[changed]` for changes in existing functionality.
- `[deprecated]` for once-stable features removed in upcoming releases.
- `[removed]` for deprecated features removed in this release.
- `[fixed]` for any bug fixes.
- `[security]` to invite users to upgrade in case of vulnerabilities.
## [Unreleased]
- `[changed]` Improve type narrowing for `is_ok` and `is_err` type guards by
replacing `typing.TypeGuard` with `typing.TypeIs` (#193)
## [0.17.0] - 2024-06-02
- `[added]` Add `inspect()` and `inspect_err()` methods (#185)
## [0.16.1] - 2024-02-29
- `[fixed]` PyPI not showing description (#176)
## [0.16.0] - 2023-12-23
- `[added]` Add `map_async` for async functions (#165)
- `[fixed]` Add `do_async()` to handle edge case in `do()` involving multiple inlined awaits (#149)
- `[added]` Add support for Python 3.12 (#157)
## [0.15.0] - 2023-12-04
- `[added]` Add `do` function to support Haskell-style do-notation (#149)
## [0.14.0] - 2023-11-10
- `[added]` `is_ok` and `is_err` type guard functions as alternatives to `isinstance` checks (#69)
- `[added]` Add `and_then_async` for async functions (#148)
## [0.13.1] - 2023-07-19
- `[fixed]` Use `self._value` instead of deprecated `self.value` in `Err.expect` and `Err.unwrap` to avoid raising a warning (#133)
## [0.13.0] - 2023-07-15
- `[changed]` Include captured `Err` value when `expect` and `unwrap` are called and an `UnwrapError` is raised (#98, #132)
## [0.12.0] - 2023-06-11
- `[removed]` Drop support for Python 3.7 (#126)
- `[fixed]` Pattern matching deprecation warning (#128)
- `[changed]` Minor internal implementation details (#129, #130)
## [0.11.0] - 2023-06-11
- `[changed]` `Ok` now requires an explicit value during instantiation. Please
check out [MIGRATING.md], it will guide you through the necessary change in
your codebase.
- `[deprecated]` `value` property to access the inner value (#37, #121)
- `[added]` `ok_value` and `err_value` to access the inner value more safely (#37, #121)
## [0.10.0] - 2023-04-29
- `[fixed]` Make python version check PEP 484 compliant (#118)
- `[added]` `as_async_result` decorator to turn regular async functions into
`Result` returning ones (#116)
## [0.9.0] - 2022-12-09
- `[added]` Implement `unwrap_or_raise` (#95)
- `[added]` Add support for Python 3.11 (#107)
- `[changed]` Narrowing of return types on methods of `Err` and `Ok`. (#106)
- `[fixed]` Fix failing type inference for `Result.map` and similar method
unions (#106)
## [0.8.0] - 2022-04-17
- `[added]` `as_result` decorator to turn regular functions into
`Result` returning ones (#33, 71)
- `[removed]` Drop support for Python 3.6 (#49)
- `[added]` Implement `unwrap_or_else` (#74), `and_then` (#90) and `or_else` (#90)
## [0.7.0] - 2021-11-19
- `[removed]` Drop support for Python 3.5 (#34)
- `[added]` Add support for Python 3.9 and 3.10 (#50)
- `[changed]` Make the `Ok` type covariant in regard to its wrapped type `T`.
Likewise for `Err` in regard to `E`. This should result in more intuitive
type checking behaviour. For instance, `Err[TypeError]` will get recognized
as a subtype of `Err[Exception]` by type checkers. See [PEP 438] for a
detailed explanation of covariance and its implications.
- `[added]` Add support for Python 3.10 pattern matching (#47)
- `[changed]` `Ok` and `Err` now define `__slots__` to save memory (#55, #58)
- `[changed]` The generic type of `UnwrapError.result` now explicitly specifies `Any` (#67)
[PEP 438]: https://www.python.org/dev/peps/pep-0483/#covariance-and-contravariance
## [0.6.0] - 2021-03-17
**IMPORTANT:** This release a big API refactoring to make the API more type
safe. Unfortunately this means some breaking changes. Please check out
[MIGRATING.md], it will guide you through the necessary changes in your
codebase.
- [changed] Split result type into `Ok` and `Err` classes (#17, #27)
- [deprecated] Python 3.4 support is deprecated and will be removed in the next
release
## [0.5.0] - 2020-03-03
- [added] Implement `map`, `map_err`, `map_or` and `map_or_else` (#19)
- [added] Add `unwrap_err` and `expect_err` methods (#26)
- [changed] Type annotations: Change parameter order
from `Result[E, T]` to `Result[T, E]` to match Rust/OCaml/F# (#7)
## [0.4.1] - 2020-02-17
- [added] Add `py.typed` for PEP561 package compliance (#16)
## [0.4.0] - 2019-04-17
- [added] Add `unwrap`, `unwrap_or` and `expect` (#9)
- [removed] Drop support for Python 2 and 3.3
- [changed] Only install typing dependency for Python <3.5
## [0.3.0] - 2017-07-12
- [added] This library is now fully type annotated (#4, thanks @tyehle)
- [added] Implementations for `__ne__`, `__hash__` and `__repr__`
- [deprecated] Python 2 support is deprecated and will be removed in the 0.4 release
## [0.2.2] - 2016-09-21
- [added] `__eq__` magic method
## [0.2.0] - 2016-05-05
- [added] Convenience default: `Ok()` == `Ok(True)`
## [0.1.1] - 2015-12-14
- [fixed] Import bugfix
## [0.1.0] - 2015-12-14
- Initial version
[MIGRATING.md]: https://github.com/rustedpy/result/blob/main/MIGRATING.md
[Unreleased]: https://github.com/rustedpy/result/compare/v0.17.0...HEAD
[0.17.0]: https://github.com/rustedpy/result/compare/v0.16.1...v0.17.0
[0.16.1]: https://github.com/rustedpy/result/compare/v0.16.0...v0.16.1
[0.16.0]: https://github.com/rustedpy/result/compare/v0.15.0...v0.16.0
[0.15.0]: https://github.com/rustedpy/result/compare/v0.14.0...v0.15.0
[0.14.0]: https://github.com/rustedpy/result/compare/v0.13.1...v0.14.0
[0.13.1]: https://github.com/rustedpy/result/compare/v0.13.0...v0.13.1
[0.13.0]: https://github.com/rustedpy/result/compare/v0.12.0...v0.13.0
[0.12.0]: https://github.com/rustedpy/result/compare/v0.11.0...v0.12.0
[0.11.0]: https://github.com/rustedpy/result/compare/v0.10.0...v0.11.0
[0.10.0]: https://github.com/rustedpy/result/compare/v0.9.0...v0.10.0
[0.9.0]: https://github.com/rustedpy/result/compare/v0.8.0...v0.9.0
[0.8.0]: https://github.com/rustedpy/result/compare/v0.7.0...v0.8.0
[0.7.0]: https://github.com/rustedpy/result/compare/v0.6.0...v0.7.0
[0.6.0]: https://github.com/rustedpy/result/compare/v0.5.0...v0.6.0
[0.5.0]: https://github.com/rustedpy/result/compare/v0.4.1...v0.5.0
[0.4.1]: https://github.com/rustedpy/result/compare/v0.4.0...v0.4.1
[0.4.0]: https://github.com/rustedpy/result/compare/v0.3.0...v0.4.0
[0.3.0]: https://github.com/rustedpy/result/compare/v0.2.2...v0.3.0
[0.2.2]: https://github.com/rustedpy/result/compare/v0.2.1...v0.2.2
[0.2.1]: https://github.com/rustedpy/result/compare/v0.2.0...v0.2.1
[0.2.0]: https://github.com/rustedpy/result/compare/v0.1.1...v0.2.0
[0.1.1]: https://github.com/rustedpy/result/compare/v0.1.0...v0.1.1
[0.1.0]: https://github.com/rustedpy/result/compare/3ca7d83...v0.1.0

19
app/external/result_type/LICENSE vendored Normal file
View file

@ -0,0 +1,19 @@
Copyright (C) 2015-2020 Danilo Bargen and contributors
Permission is hereby granted, free of charge, to any person obtaining a copy of
this software and associated documentation files (the "Software"), to deal in
the Software without restriction, including without limitation the rights to
use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies
of the Software, and to permit persons to whom the Software is furnished to do
so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.

484
app/external/result_type/README.md vendored Normal file
View file

@ -0,0 +1,484 @@
# Result
[![GitHub Workflow Status (branch)](https://img.shields.io/github/actions/workflow/status/rustedpy/result/ci.yml?branch=main)](https://github.com/rustedpy/result/actions/workflows/ci.yml?query=branch%3Amain)
[![Coverage](https://codecov.io/gh/rustedpy/result/branch/main/graph/badge.svg)](https://codecov.io/gh/rustedpy/result)
A simple Result type for Python 3 [inspired by
Rust](https://doc.rust-lang.org/std/result/), fully type annotated.
## Installation
Latest release:
``` sh
$ pip install result
```
Latest GitHub `main` branch version:
``` sh
$ pip install git+https://github.com/rustedpy/result
```
## Summary
The idea is that a result value can be either `Ok(value)` or
`Err(error)`, with a way to differentiate between the two. `Ok` and
`Err` are both classes encapsulating an arbitrary value. `Result[T, E]`
is a generic type alias for `typing.Union[Ok[T], Err[E]]`. It will
change code like this:
``` python
def get_user_by_email(email: str) -> Tuple[Optional[User], Optional[str]]:
"""
Return the user instance or an error message.
"""
if not user_exists(email):
return None, 'User does not exist'
if not user_active(email):
return None, 'User is inactive'
user = get_user(email)
return user, None
user, reason = get_user_by_email('ueli@example.com')
if user is None:
raise RuntimeError('Could not fetch user: %s' % reason)
else:
do_something(user)
```
To something like this:
``` python
from result import Ok, Err, Result, is_ok, is_err
def get_user_by_email(email: str) -> Result[User, str]:
"""
Return the user instance or an error message.
"""
if not user_exists(email):
return Err('User does not exist')
if not user_active(email):
return Err('User is inactive')
user = get_user(email)
return Ok(user)
user_result = get_user_by_email(email)
if is_ok(user_result):
# type(user_result.ok_value) == User
do_something(user_result.ok_value)
else:
# type(user_result.err_value) == str
raise RuntimeError('Could not fetch user: %s' % user_result.err_value)
```
Note that `.ok_value` exists only on an instance of `Ok` and
`.err_value` exists only on an instance of `Err`.
And if you're using python version `3.10` or later, you can use the
elegant `match` statement as well:
``` python
from result import Result, Ok, Err
def divide(a: int, b: int) -> Result[int, str]:
if b == 0:
return Err("Cannot divide by zero")
return Ok(a // b)
values = [(10, 0), (10, 5)]
for a, b in values:
match divide(a, b):
case Ok(value):
print(f"{a} // {b} == {value}")
case Err(e):
print(e)
```
Not all methods
(<https://doc.rust-lang.org/std/result/enum.Result.html>) have been
implemented, only the ones that make sense in the Python context.
All of this in a package allowing easier handling of values that can
be OK or not, without resorting to custom exceptions.
## API
Auto generated API docs are also available at
[./docs/README.md](./docs/README.md).
Creating an instance:
``` python
>>> from result import Ok, Err
>>> res1 = Ok('yay')
>>> res2 = Err('nay')
```
Checking whether a result is `Ok` or `Err`:
``` python
if is_err(result):
raise RuntimeError(result.err_value)
do_something(result.ok_value)
```
or
``` python
if is_ok(result):
do_something(result.ok_value)
else:
raise RuntimeError(result.err_value)
```
Alternatively, `isinstance` can be used (interchangeably to type guard functions
`is_ok` and `is_err`). However, relying on `isinstance` may result in code that
is slightly less readable and less concise:
``` python
if isinstance(result, Err):
raise RuntimeError(result.err_value)
do_something(result.ok_value)
```
You can also check if an object is `Ok` or `Err` by using the `OkErr`
type. Please note that this type is designed purely for convenience, and
should not be used for anything else. Using `(Ok, Err)` also works fine:
``` python
>>> res1 = Ok('yay')
>>> res2 = Err('nay')
>>> isinstance(res1, OkErr)
True
>>> isinstance(res2, OkErr)
True
>>> isinstance(1, OkErr)
False
>>> isinstance(res1, (Ok, Err))
True
```
Convert a `Result` to the value or `None`:
``` python
>>> res1 = Ok('yay')
>>> res2 = Err('nay')
>>> res1.ok()
'yay'
>>> res2.ok()
None
```
Convert a `Result` to the error or `None`:
``` python
>>> res1 = Ok('yay')
>>> res2 = Err('nay')
>>> res1.err()
None
>>> res2.err()
'nay'
```
Access the value directly, without any other checks:
``` python
>>> res1 = Ok('yay')
>>> res2 = Err('nay')
>>> res1.ok_value
'yay'
>>> res2.err_value
'nay'
```
Note that this is a property, you cannot assign to it. Results are
immutable.
When the value inside is irrelevant, we suggest using `None` or a
`bool`, but you're free to use any value you think works best. An
instance of a `Result` (`Ok` or `Err`) must always contain something. If
you're looking for a type that might contain a value you may be
interested in a [maybe](https://github.com/rustedpy/maybe).
The `unwrap` method returns the value if `Ok` and `unwrap_err` method
returns the error value if `Err`, otherwise it raises an `UnwrapError`:
``` python
>>> res1 = Ok('yay')
>>> res2 = Err('nay')
>>> res1.unwrap()
'yay'
>>> res2.unwrap()
Traceback (most recent call last):
File "<stdin>", line 1, in <module>
File "C:\project\result\result.py", line 107, in unwrap
return self.expect("Called `Result.unwrap()` on an `Err` value")
File "C:\project\result\result.py", line 101, in expect
raise UnwrapError(message)
result.result.UnwrapError: Called `Result.unwrap()` on an `Err` value
>>> res1.unwrap_err()
Traceback (most recent call last):
...
>>>res2.unwrap_err()
'nay'
```
A custom error message can be displayed instead by using `expect` and
`expect_err`:
``` python
>>> res1 = Ok('yay')
>>> res2 = Err('nay')
>>> res1.expect('not ok')
'yay'
>>> res2.expect('not ok')
Traceback (most recent call last):
File "<stdin>", line 1, in <module>
File "C:\project\result\result.py", line 101, in expect
raise UnwrapError(message)
result.result.UnwrapError: not ok
>>> res1.expect_err('not err')
Traceback (most recent call last):
...
>>> res2.expect_err('not err')
'nay'
```
A default value can be returned instead by using `unwrap_or` or
`unwrap_or_else`:
``` python
>>> res1 = Ok('yay')
>>> res2 = Err('nay')
>>> res1.unwrap_or('default')
'yay'
>>> res2.unwrap_or('default')
'default'
>>> res1.unwrap_or_else(str.upper)
'yay'
>>> res2.unwrap_or_else(str.upper)
'NAY'
```
The `unwrap` method will raised an `UnwrapError`. A custom exception can
be raised by using the `unwrap_or_raise` method instead:
``` python
>>> res1 = Ok('yay')
>>> res2 = Err('nay')
>>> res1.unwrap_or_raise(ValueError)
'yay'
>>> res2.unwrap_or_raise(ValueError)
ValueError: nay
```
Values and errors can be mapped using `map`, `map_or`, `map_or_else` and
`map_err`:
``` python
>>> Ok(1).map(lambda x: x + 1)
Ok(2)
>>> Err('nay').map(lambda x: x + 1)
Err('nay')
>>> Ok(1).map_or(-1, lambda x: x + 1)
2
>>> Err(1).map_or(-1, lambda x: x + 1)
-1
>>> Ok(1).map_or_else(lambda: 3, lambda x: x + 1)
2
>>> Err('nay').map_or_else(lambda: 3, lambda x: x + 1)
3
>>> Ok(1).map_err(lambda x: x + 1)
Ok(1)
>>> Err(1).map_err(lambda x: x + 1)
Err(2)
```
To save memory, both the `Ok` and `Err` classes are slotted, i.e. they
define `__slots__`. This means assigning arbitrary attributes to
instances will raise `AttributeError`.
### `as_result` Decorator
The `as_result()` decorator can be used to quickly turn normal
functions into `Result` returning ones by specifying one or more
exception types:
``` python
@as_result(ValueError, IndexError)
def f(value: int) -> int:
if value == 0:
raise ValueError # becomes Err
elif value == 1:
raise IndexError # becomes Err
elif value == 2:
raise KeyError # raises Exception
else:
return value # becomes Ok
res = f(0) # Err[ValueError()]
res = f(1) # Err[IndexError()]
res = f(2) # raises KeyError
res = f(3) # Ok[3]
```
`Exception` (or even `BaseException`) can be specified to create a
catch all `Result` return type. This is effectively the same as `try`
followed by `except Exception`, which is not considered good practice in
most scenarios, and hence this requires explicit opt-in.
Since `as_result` is a regular decorator, it can be used to wrap
existing functions (also from other libraries), albeit with a slightly
unconventional syntax (without the usual `@`):
``` python
import third_party
x = third_party.do_something(...) # could raise; who knows?
safe_do_something = as_result(Exception)(third_party.do_something)
res = safe_do_something(...) # Ok(...) or Err(...)
if is_ok(res):
print(res.ok_value)
```
### Do notation
Do notation is syntactic sugar for a sequence of `and_then()` calls.
Much like the equivalent in Rust or Haskell, but with different syntax.
Instead of `x <- Ok(1)` we write `for x in Ok(1)`. Since the syntax is
generator-based, the final result must be the first line, not the last.
``` python
final_result: Result[int, str] = do(
Ok(x + y)
for x in Ok(1)
for y in Ok(2)
)
```
Note that if you exclude the type annotation,
`final_result: Result[float, int] = ...`, your type checker may be
unable to infer the return type. To avoid an errors or warnings from
your type checker, you should add a type hint when using the `do`
function.
This is similar to Rust's [m!
macro](https://docs.rs/do-notation/latest/do_notation/):
``` rust
use do_notation::m;
let r = m! {
x <- Some(1);
y <- Some(2);
Some(x + y)
};
```
Note that if your do statement has multiple <span
class="title-ref">for\`s, you can access an identifier bound in a
previous \`for</span>. Example:
``` python
my_result: Result[int, str] = do(
f(x, y, z)
for x in get_x()
for y in calculate_y_from_x(x)
for z in calculate_z_from_x_y(x, y)
)
```
You can use `do()` with awaited values as follows:
``` python
async def process_data(data) -> Result[int, str]:
res1 = await get_result_1(data)
res2 = await get_result_2(data)
return do(
Ok(x + y)
for x in res1
for y in res2
)
```
However, if you want to await something inside the expression, use
`do_async()`:
``` python
async def process_data(data) -> Result[int, str]:
return do_async(
Ok(x + y)
for x in await get_result_1(data)
for y in await get_result_2(data)
)
```
Troubleshooting `do()` calls:
``` python
TypeError("Got async_generator but expected generator")
```
Sometimes regular `do()` can handle async values, but this error means
you have hit a case where it does not. You should use `do_async()` here
instead.
## Contributing
These steps should work on any Unix-based system (Linux, macOS, etc) with Python
and `make` installed. On Windows, you will need to refer to the Python
documentation (linked below) and reference the `Makefile` for commands to run
from the non-unix shell you're using on Windows.
1. Setup and activate a virtual environment. See [Python docs][pydocs-venv] for more
information about virtual environments and setup.
2. Run `make install` to install dependencies
3. Switch to a new git branch and make your changes
4. Test your changes:
- `make test`
- `make lint`
- You can also start a Python REPL and import `result`
5. Update documentation
- Edit any relevant docstrings, markdown files
- Run `make docs`
6. Add an entry to the [changelog](./CHANGELOG.md)
5. Git commit all your changes and create a new PR.
[pydocs-venv]: https://docs.python.org/3/library/venv.html
## FAQ
- **Why should I use the `is_ok` (`is_err`) type guard function over the `is_ok` (`is_err`) method?**
As you can see in the following example, MyPy can only narrow the type correctly
while using the type guard **functions**:
```python
result: Result[int, str]
if is_ok(result):
reveal_type(result) # "result.result.Ok[builtins.int]"
else:
reveal_type(result) # "result.result.Err[builtins.str]"
if result.is_ok():
reveal_type(result) # "Union[result.result.Ok[builtins.int], result.result.Err[builtins.str]]"
else:
reveal_type(result) # "Union[result.result.Ok[builtins.int], result.result.Err[builtins.str]]"
```
- **Why do I get the "Cannot infer type argument" error with MyPy?**
There is [a bug in MyPy](https://github.com/python/mypy/issues/230)
which can be triggered in some scenarios. Using `if isinstance(res, Ok)`
instead of `if res.is_ok()` will help in some cases. Otherwise using
[one of these
workarounds](https://github.com/python/mypy/issues/3889#issuecomment-325997911)
can help.
## Related Projects
- [dry-python/returns: Make your functions return something meaningful, typed, and safe!](https://github.com/dry-python/returns)
- [alexandermalyga/poltergeist: Rust-like error handling in Python, with type-safety in mind.](https://github.com/alexandermalyga/poltergeist)
## License
MIT License

4
app/external/result_type/docs/.pages vendored Normal file
View file

@ -0,0 +1,4 @@
title: API Reference
nav:
- Overview: README.md
- ...

28
app/external/result_type/docs/README.md vendored Normal file
View file

@ -0,0 +1,28 @@
<!-- markdownlint-disable -->
# API Overview
## Modules
- [`result`](./result.md#module-result)
## Classes
- [`result.DoException`](./result.md#class-doexception): This is used to signal to `do()` that the result is an `Err`,
- [`result.Err`](./result.md#class-err): A value that signifies failure and which stores arbitrary data for the error.
- [`result.Ok`](./result.md#class-ok): A value that indicates success and which stores arbitrary data for the return value.
- [`result.UnwrapError`](./result.md#class-unwraperror): Exception raised from ``.unwrap_<...>`` and ``.expect_<...>`` calls.
## Functions
- [`result.as_async_result`](./result.md#function-as_async_result): Make a decorator to turn an async function into one that returns a ``Result``.
- [`result.as_result`](./result.md#function-as_result): Make a decorator to turn a function into one that returns a ``Result``.
- [`result.do`](./result.md#function-do): Do notation for Result (syntactic sugar for sequence of `and_then()` calls).
- [`result.do_async`](./result.md#function-do_async): Async version of do. Example:
- [`result.is_err`](./result.md#function-is_err): A type guard to check if a result is an Err
- [`result.is_ok`](./result.md#function-is_ok): A type guard to check if a result is an Ok
---
_This file was automatically generated via [lazydocs](https://github.com/ml-tooling/lazydocs)._

827
app/external/result_type/docs/result.md vendored Normal file
View file

@ -0,0 +1,827 @@
<!-- markdownlint-disable -->
<a href="https://github.com/rustedpy/result/blob/main/src/result/result.py#L0"><img align="right" style="float:right;" src="https://img.shields.io/badge/-source-cccccc?style=flat-square"></a>
# <kbd>module</kbd> `result`
**Global Variables**
---------------
- **OkErr**
---
<a href="https://github.com/rustedpy/result/blob/main/src/result/result.py#L467"><img align="right" style="float:right;" src="https://img.shields.io/badge/-source-cccccc?style=flat-square"></a>
## <kbd>function</kbd> `as_result`
```python
as_result(
*exceptions: 'Type[TBE]'
) → Callable[[Callable[P, R]], Callable[P, Result[R, TBE]]]
```
Make a decorator to turn a function into one that returns a ``Result``.
Regular return values are turned into ``Ok(return_value)``. Raised exceptions of the specified exception type(s) are turned into ``Err(exc)``.
---
<a href="https://github.com/rustedpy/result/blob/main/src/result/result.py#L499"><img align="right" style="float:right;" src="https://img.shields.io/badge/-source-cccccc?style=flat-square"></a>
## <kbd>function</kbd> `as_async_result`
```python
as_async_result(
*exceptions: 'Type[TBE]'
) → Callable[[Callable[P, Awaitable[R]]], Callable[P, Awaitable[Result[R, TBE]]]]
```
Make a decorator to turn an async function into one that returns a ``Result``. Regular return values are turned into ``Ok(return_value)``. Raised exceptions of the specified exception type(s) are turned into ``Err(exc)``.
---
<a href="https://github.com/rustedpy/result/blob/main/src/result/result.py#L532"><img align="right" style="float:right;" src="https://img.shields.io/badge/-source-cccccc?style=flat-square"></a>
## <kbd>function</kbd> `is_ok`
```python
is_ok(result: 'Result[T, E]') → TypeIs[Ok[T]]
```
A type guard to check if a result is an Ok
Usage:
``` python
r: Result[int, str] = get_a_result()
if is_ok(r):
r # r is of type Ok[int]
elif is_err(r):
r # r is of type Err[str]
```
---
<a href="https://github.com/rustedpy/result/blob/main/src/result/result.py#L549"><img align="right" style="float:right;" src="https://img.shields.io/badge/-source-cccccc?style=flat-square"></a>
## <kbd>function</kbd> `is_err`
```python
is_err(result: 'Result[T, E]') → TypeIs[Err[E]]
```
A type guard to check if a result is an Err
Usage:
``` python
r: Result[int, str] = get_a_result()
if is_ok(r):
r # r is of type Ok[int]
elif is_err(r):
r # r is of type Err[str]
```
---
<a href="https://github.com/rustedpy/result/blob/main/src/result/result.py#L566"><img align="right" style="float:right;" src="https://img.shields.io/badge/-source-cccccc?style=flat-square"></a>
## <kbd>function</kbd> `do`
```python
do(gen: 'Generator[Result[T, E], None, None]') → Result[T, E]
```
Do notation for Result (syntactic sugar for sequence of `and_then()` calls).
Usage:
``` rust
// This is similar to
use do_notation::m;
let final_result = m! {
x <- Ok("hello");
y <- Ok(True);
Ok(len(x) + int(y) + 0.5)
};
```
``` rust
final_result: Result[float, int] = do(
Ok(len(x) + int(y) + 0.5)
for x in Ok("hello")
for y in Ok(True)
)
```
NOTE: If you exclude the type annotation e.g. `Result[float, int]` your type checker might be unable to infer the return type. To avoid an error, you might need to help it with the type hint.
---
<a href="https://github.com/rustedpy/result/blob/main/src/result/result.py#L611"><img align="right" style="float:right;" src="https://img.shields.io/badge/-source-cccccc?style=flat-square"></a>
## <kbd>function</kbd> `do_async`
```python
do_async(
gen: 'Union[Generator[Result[T, E], None, None], AsyncGenerator[Result[T, E], None]]'
) → Result[T, E]
```
Async version of do. Example:
``` python
final_result: Result[float, int] = await do_async(
Ok(len(x) + int(y) + z)
for x in await get_async_result_1()
for y in await get_async_result_2()
for z in get_sync_result_3()
)
```
NOTE: Python makes generators async in a counter-intuitive way.
``` python
# This is a regular generator:
async def foo(): ...
do(Ok(1) for x in await foo())
```
``` python
# But this is an async generator:
async def foo(): ...
async def bar(): ...
do(
Ok(1)
for x in await foo()
for y in await bar()
)
```
We let users try to use regular `do()`, which works in some cases of awaiting async values. If we hit a case like above, we raise an exception telling the user to use `do_async()` instead. See `do()`.
However, for better usability, it's better for `do_async()` to also accept regular generators, as you get in the first case:
``` python
async def foo(): ...
do(Ok(1) for x in await foo())
```
Furthermore, neither mypy nor pyright can infer that the second case is actually an async generator, so we cannot annotate `do_async()` as accepting only an async generator. This is additional motivation to accept either.
---
<a href="https://github.com/rustedpy/result/blob/main/src/result/result.py#L40"><img align="right" style="float:right;" src="https://img.shields.io/badge/-source-cccccc?style=flat-square"></a>
## <kbd>class</kbd> `Ok`
A value that indicates success and which stores arbitrary data for the return value.
<a href="https://github.com/rustedpy/result/blob/main/src/result/result.py#L51"><img align="right" style="float:right;" src="https://img.shields.io/badge/-source-cccccc?style=flat-square"></a>
### <kbd>method</kbd> `__init__`
```python
__init__(value: 'T') → None
```
---
#### <kbd>property</kbd> ok_value
Return the inner value.
---
#### <kbd>property</kbd> value
Return the inner value.
@deprecated Use `ok_value` or `err_value` instead. This method will be removed in a future version.
---
<a href="https://github.com/rustedpy/result/blob/main/src/result/result.py#L185"><img align="right" style="float:right;" src="https://img.shields.io/badge/-source-cccccc?style=flat-square"></a>
### <kbd>method</kbd> `and_then`
```python
and_then(op: 'Callable[[T], Result[U, E]]') → Result[U, E]
```
The contained result is `Ok`, so return the result of `op` with the original value passed in
---
<a href="https://github.com/rustedpy/result/blob/main/src/result/result.py#L192"><img align="right" style="float:right;" src="https://img.shields.io/badge/-source-cccccc?style=flat-square"></a>
### <kbd>method</kbd> `and_then_async`
```python
and_then_async(op: 'Callable[[T], Awaitable[Result[U, E]]]') → Result[U, E]
```
The contained result is `Ok`, so return the result of `op` with the original value passed in
---
<a href="https://github.com/rustedpy/result/blob/main/src/result/result.py#L78"><img align="right" style="float:right;" src="https://img.shields.io/badge/-source-cccccc?style=flat-square"></a>
### <kbd>method</kbd> `err`
```python
err() → None
```
Return `None`.
---
<a href="https://github.com/rustedpy/result/blob/main/src/result/result.py#L107"><img align="right" style="float:right;" src="https://img.shields.io/badge/-source-cccccc?style=flat-square"></a>
### <kbd>method</kbd> `expect`
```python
expect(_message: 'str') → T
```
Return the value.
---
<a href="https://github.com/rustedpy/result/blob/main/src/result/result.py#L113"><img align="right" style="float:right;" src="https://img.shields.io/badge/-source-cccccc?style=flat-square"></a>
### <kbd>method</kbd> `expect_err`
```python
expect_err(message: 'str') → NoReturn
```
Raise an UnwrapError since this type is `Ok`
---
<a href="https://github.com/rustedpy/result/blob/main/src/result/result.py#L207"><img align="right" style="float:right;" src="https://img.shields.io/badge/-source-cccccc?style=flat-square"></a>
### <kbd>method</kbd> `inspect`
```python
inspect(op: 'Callable[[T], Any]') → Result[T, E]
```
Calls a function with the contained value if `Ok`. Returns the original result.
---
<a href="https://github.com/rustedpy/result/blob/main/src/result/result.py#L214"><img align="right" style="float:right;" src="https://img.shields.io/badge/-source-cccccc?style=flat-square"></a>
### <kbd>method</kbd> `inspect_err`
```python
inspect_err(op: 'Callable[[E], Any]') → Result[T, E]
```
Calls a function with the contained value if `Err`. Returns the original result.
---
<a href="https://github.com/rustedpy/result/blob/main/src/result/result.py#L69"><img align="right" style="float:right;" src="https://img.shields.io/badge/-source-cccccc?style=flat-square"></a>
### <kbd>method</kbd> `is_err`
```python
is_err() → Literal[False]
```
---
<a href="https://github.com/rustedpy/result/blob/main/src/result/result.py#L66"><img align="right" style="float:right;" src="https://img.shields.io/badge/-source-cccccc?style=flat-square"></a>
### <kbd>method</kbd> `is_ok`
```python
is_ok() → Literal[True]
```
---
<a href="https://github.com/rustedpy/result/blob/main/src/result/result.py#L149"><img align="right" style="float:right;" src="https://img.shields.io/badge/-source-cccccc?style=flat-square"></a>
### <kbd>method</kbd> `map`
```python
map(op: 'Callable[[T], U]') → Ok[U]
```
The contained result is `Ok`, so return `Ok` with original value mapped to a new value using the passed in function.
---
<a href="https://github.com/rustedpy/result/blob/main/src/result/result.py#L156"><img align="right" style="float:right;" src="https://img.shields.io/badge/-source-cccccc?style=flat-square"></a>
### <kbd>method</kbd> `map_async`
```python
map_async(op: 'Callable[[T], Awaitable[U]]') → Ok[U]
```
The contained result is `Ok`, so return the result of `op` with the original value passed in
---
<a href="https://github.com/rustedpy/result/blob/main/src/result/result.py#L179"><img align="right" style="float:right;" src="https://img.shields.io/badge/-source-cccccc?style=flat-square"></a>
### <kbd>method</kbd> `map_err`
```python
map_err(op: 'object') → Ok[T]
```
The contained result is `Ok`, so return `Ok` with the original value
---
<a href="https://github.com/rustedpy/result/blob/main/src/result/result.py#L165"><img align="right" style="float:right;" src="https://img.shields.io/badge/-source-cccccc?style=flat-square"></a>
### <kbd>method</kbd> `map_or`
```python
map_or(default: 'object', op: 'Callable[[T], U]') → U
```
The contained result is `Ok`, so return the original value mapped to a new value using the passed in function.
---
<a href="https://github.com/rustedpy/result/blob/main/src/result/result.py#L172"><img align="right" style="float:right;" src="https://img.shields.io/badge/-source-cccccc?style=flat-square"></a>
### <kbd>method</kbd> `map_or_else`
```python
map_or_else(default_op: 'object', op: 'Callable[[T], U]') → U
```
The contained result is `Ok`, so return original value mapped to a new value using the passed in `op` function.
---
<a href="https://github.com/rustedpy/result/blob/main/src/result/result.py#L72"><img align="right" style="float:right;" src="https://img.shields.io/badge/-source-cccccc?style=flat-square"></a>
### <kbd>method</kbd> `ok`
```python
ok() → T
```
Return the value.
---
<a href="https://github.com/rustedpy/result/blob/main/src/result/result.py#L201"><img align="right" style="float:right;" src="https://img.shields.io/badge/-source-cccccc?style=flat-square"></a>
### <kbd>method</kbd> `or_else`
```python
or_else(op: 'object') → Ok[T]
```
The contained result is `Ok`, so return `Ok` with the original value
---
<a href="https://github.com/rustedpy/result/blob/main/src/result/result.py#L119"><img align="right" style="float:right;" src="https://img.shields.io/badge/-source-cccccc?style=flat-square"></a>
### <kbd>method</kbd> `unwrap`
```python
unwrap() → T
```
Return the value.
---
<a href="https://github.com/rustedpy/result/blob/main/src/result/result.py#L125"><img align="right" style="float:right;" src="https://img.shields.io/badge/-source-cccccc?style=flat-square"></a>
### <kbd>method</kbd> `unwrap_err`
```python
unwrap_err() → NoReturn
```
Raise an UnwrapError since this type is `Ok`
---
<a href="https://github.com/rustedpy/result/blob/main/src/result/result.py#L131"><img align="right" style="float:right;" src="https://img.shields.io/badge/-source-cccccc?style=flat-square"></a>
### <kbd>method</kbd> `unwrap_or`
```python
unwrap_or(_default: 'U') → T
```
Return the value.
---
<a href="https://github.com/rustedpy/result/blob/main/src/result/result.py#L137"><img align="right" style="float:right;" src="https://img.shields.io/badge/-source-cccccc?style=flat-square"></a>
### <kbd>method</kbd> `unwrap_or_else`
```python
unwrap_or_else(op: 'object') → T
```
Return the value.
---
<a href="https://github.com/rustedpy/result/blob/main/src/result/result.py#L143"><img align="right" style="float:right;" src="https://img.shields.io/badge/-source-cccccc?style=flat-square"></a>
### <kbd>method</kbd> `unwrap_or_raise`
```python
unwrap_or_raise(e: 'object') → T
```
Return the value.
---
<a href="https://github.com/rustedpy/result/blob/main/src/result/result.py#L221"><img align="right" style="float:right;" src="https://img.shields.io/badge/-source-cccccc?style=flat-square"></a>
## <kbd>class</kbd> `DoException`
This is used to signal to `do()` that the result is an `Err`, which short-circuits the generator and returns that Err. Using this exception for control flow in `do()` allows us to simulate `and_then()` in the Err case: namely, we don't call `op`, we just return `self` (the Err).
<a href="https://github.com/rustedpy/result/blob/main/src/result/result.py#L230"><img align="right" style="float:right;" src="https://img.shields.io/badge/-source-cccccc?style=flat-square"></a>
### <kbd>method</kbd> `__init__`
```python
__init__(err: 'Err[E]') → None
```
---
<a href="https://github.com/rustedpy/result/blob/main/src/result/result.py#L234"><img align="right" style="float:right;" src="https://img.shields.io/badge/-source-cccccc?style=flat-square"></a>
## <kbd>class</kbd> `Err`
A value that signifies failure and which stores arbitrary data for the error.
<a href="https://github.com/rustedpy/result/blob/main/src/result/result.py#L250"><img align="right" style="float:right;" src="https://img.shields.io/badge/-source-cccccc?style=flat-square"></a>
### <kbd>method</kbd> `__init__`
```python
__init__(value: 'E') → None
```
---
#### <kbd>property</kbd> err_value
Return the inner value.
---
#### <kbd>property</kbd> value
Return the inner value.
@deprecated Use `ok_value` or `err_value` instead. This method will be removed in a future version.
---
<a href="https://github.com/rustedpy/result/blob/main/src/result/result.py#L393"><img align="right" style="float:right;" src="https://img.shields.io/badge/-source-cccccc?style=flat-square"></a>
### <kbd>method</kbd> `and_then`
```python
and_then(op: 'object') → Err[E]
```
The contained result is `Err`, so return `Err` with the original value
---
<a href="https://github.com/rustedpy/result/blob/main/src/result/result.py#L399"><img align="right" style="float:right;" src="https://img.shields.io/badge/-source-cccccc?style=flat-square"></a>
### <kbd>method</kbd> `and_then_async`
```python
and_then_async(op: 'object') → Err[E]
```
The contained result is `Err`, so return `Err` with the original value
---
<a href="https://github.com/rustedpy/result/blob/main/src/result/result.py#L277"><img align="right" style="float:right;" src="https://img.shields.io/badge/-source-cccccc?style=flat-square"></a>
### <kbd>method</kbd> `err`
```python
err() → E
```
Return the error.
---
<a href="https://github.com/rustedpy/result/blob/main/src/result/result.py#L306"><img align="right" style="float:right;" src="https://img.shields.io/badge/-source-cccccc?style=flat-square"></a>
### <kbd>method</kbd> `expect`
```python
expect(message: 'str') → NoReturn
```
Raises an `UnwrapError`.
---
<a href="https://github.com/rustedpy/result/blob/main/src/result/result.py#L318"><img align="right" style="float:right;" src="https://img.shields.io/badge/-source-cccccc?style=flat-square"></a>
### <kbd>method</kbd> `expect_err`
```python
expect_err(_message: 'str') → E
```
Return the inner value
---
<a href="https://github.com/rustedpy/result/blob/main/src/result/result.py#L412"><img align="right" style="float:right;" src="https://img.shields.io/badge/-source-cccccc?style=flat-square"></a>
### <kbd>method</kbd> `inspect`
```python
inspect(op: 'Callable[[T], Any]') → Result[T, E]
```
Calls a function with the contained value if `Ok`. Returns the original result.
---
<a href="https://github.com/rustedpy/result/blob/main/src/result/result.py#L418"><img align="right" style="float:right;" src="https://img.shields.io/badge/-source-cccccc?style=flat-square"></a>
### <kbd>method</kbd> `inspect_err`
```python
inspect_err(op: 'Callable[[E], Any]') → Result[T, E]
```
Calls a function with the contained value if `Err`. Returns the original result.
---
<a href="https://github.com/rustedpy/result/blob/main/src/result/result.py#L268"><img align="right" style="float:right;" src="https://img.shields.io/badge/-source-cccccc?style=flat-square"></a>
### <kbd>method</kbd> `is_err`
```python
is_err() → Literal[True]
```
---
<a href="https://github.com/rustedpy/result/blob/main/src/result/result.py#L265"><img align="right" style="float:right;" src="https://img.shields.io/badge/-source-cccccc?style=flat-square"></a>
### <kbd>method</kbd> `is_ok`
```python
is_ok() → Literal[False]
```
---
<a href="https://github.com/rustedpy/result/blob/main/src/result/result.py#L361"><img align="right" style="float:right;" src="https://img.shields.io/badge/-source-cccccc?style=flat-square"></a>
### <kbd>method</kbd> `map`
```python
map(op: 'object') → Err[E]
```
Return `Err` with the same value
---
<a href="https://github.com/rustedpy/result/blob/main/src/result/result.py#L367"><img align="right" style="float:right;" src="https://img.shields.io/badge/-source-cccccc?style=flat-square"></a>
### <kbd>method</kbd> `map_async`
```python
map_async(op: 'object') → Err[E]
```
The contained result is `Ok`, so return the result of `op` with the original value passed in
---
<a href="https://github.com/rustedpy/result/blob/main/src/result/result.py#L386"><img align="right" style="float:right;" src="https://img.shields.io/badge/-source-cccccc?style=flat-square"></a>
### <kbd>method</kbd> `map_err`
```python
map_err(op: 'Callable[[E], F]') → Err[F]
```
The contained result is `Err`, so return `Err` with original error mapped to a new value using the passed in function.
---
<a href="https://github.com/rustedpy/result/blob/main/src/result/result.py#L374"><img align="right" style="float:right;" src="https://img.shields.io/badge/-source-cccccc?style=flat-square"></a>
### <kbd>method</kbd> `map_or`
```python
map_or(default: 'U', op: 'object') → U
```
Return the default value
---
<a href="https://github.com/rustedpy/result/blob/main/src/result/result.py#L380"><img align="right" style="float:right;" src="https://img.shields.io/badge/-source-cccccc?style=flat-square"></a>
### <kbd>method</kbd> `map_or_else`
```python
map_or_else(default_op: 'Callable[[], U]', op: 'object') → U
```
Return the result of the default operation
---
<a href="https://github.com/rustedpy/result/blob/main/src/result/result.py#L271"><img align="right" style="float:right;" src="https://img.shields.io/badge/-source-cccccc?style=flat-square"></a>
### <kbd>method</kbd> `ok`
```python
ok() → None
```
Return `None`.
---
<a href="https://github.com/rustedpy/result/blob/main/src/result/result.py#L405"><img align="right" style="float:right;" src="https://img.shields.io/badge/-source-cccccc?style=flat-square"></a>
### <kbd>method</kbd> `or_else`
```python
or_else(op: 'Callable[[E], Result[T, F]]') → Result[T, F]
```
The contained result is `Err`, so return the result of `op` with the original value passed in
---
<a href="https://github.com/rustedpy/result/blob/main/src/result/result.py#L324"><img align="right" style="float:right;" src="https://img.shields.io/badge/-source-cccccc?style=flat-square"></a>
### <kbd>method</kbd> `unwrap`
```python
unwrap() → NoReturn
```
Raises an `UnwrapError`.
---
<a href="https://github.com/rustedpy/result/blob/main/src/result/result.py#L336"><img align="right" style="float:right;" src="https://img.shields.io/badge/-source-cccccc?style=flat-square"></a>
### <kbd>method</kbd> `unwrap_err`
```python
unwrap_err() → E
```
Return the inner value
---
<a href="https://github.com/rustedpy/result/blob/main/src/result/result.py#L342"><img align="right" style="float:right;" src="https://img.shields.io/badge/-source-cccccc?style=flat-square"></a>
### <kbd>method</kbd> `unwrap_or`
```python
unwrap_or(default: 'U') → U
```
Return `default`.
---
<a href="https://github.com/rustedpy/result/blob/main/src/result/result.py#L348"><img align="right" style="float:right;" src="https://img.shields.io/badge/-source-cccccc?style=flat-square"></a>
### <kbd>method</kbd> `unwrap_or_else`
```python
unwrap_or_else(op: 'Callable[[E], T]') → T
```
The contained result is ``Err``, so return the result of applying ``op`` to the error value.
---
<a href="https://github.com/rustedpy/result/blob/main/src/result/result.py#L355"><img align="right" style="float:right;" src="https://img.shields.io/badge/-source-cccccc?style=flat-square"></a>
### <kbd>method</kbd> `unwrap_or_raise`
```python
unwrap_or_raise(e: 'Type[TBE]') → NoReturn
```
The contained result is ``Err``, so raise the exception with the value.
---
<a href="https://github.com/rustedpy/result/blob/main/src/result/result.py#L442"><img align="right" style="float:right;" src="https://img.shields.io/badge/-source-cccccc?style=flat-square"></a>
## <kbd>class</kbd> `UnwrapError`
Exception raised from ``.unwrap_<...>`` and ``.expect_<...>`` calls.
The original ``Result`` can be accessed via the ``.result`` attribute, but this is not intended for regular use, as type information is lost: ``UnwrapError`` doesn't know about both ``T`` and ``E``, since it's raised from ``Ok()`` or ``Err()`` which only knows about either ``T`` or ``E``, not both.
<a href="https://github.com/rustedpy/result/blob/main/src/result/result.py#L455"><img align="right" style="float:right;" src="https://img.shields.io/badge/-source-cccccc?style=flat-square"></a>
### <kbd>method</kbd> `__init__`
```python
__init__(result: 'Result[object, object]', message: 'str') → None
```
---
#### <kbd>property</kbd> result
Returns the original result.
---
_This file was automatically generated via [lazydocs](https://github.com/ml-tooling/lazydocs)._

View file

@ -0,0 +1,28 @@
from .result import (
Err,
Ok,
OkErr,
Result,
UnwrapError,
as_async_result,
as_result,
is_ok,
is_err,
do,
do_async,
)
__all__ = [
"Err",
"Ok",
"OkErr",
"Result",
"UnwrapError",
"as_async_result",
"as_result",
"is_ok",
"is_err",
"do",
"do_async",
]
__version__ = "0.18.0.dev0"

View file

View file

@ -0,0 +1,669 @@
from __future__ import annotations
import functools
import inspect
import sys
from warnings import warn
from typing import (
Any,
AsyncGenerator,
Awaitable,
Callable,
Final,
Generator,
Generic,
Iterator,
Literal,
NoReturn,
Type,
TypeVar,
Union,
)
from typing_extensions import TypeIs
if sys.version_info >= (3, 10):
from typing import ParamSpec, TypeAlias
else:
from typing_extensions import ParamSpec, TypeAlias
T = TypeVar("T", covariant=True) # Success type
E = TypeVar("E", covariant=True) # Error type
U = TypeVar("U")
F = TypeVar("F")
P = ParamSpec("P")
R = TypeVar("R")
TBE = TypeVar("TBE", bound=BaseException)
class Ok(Generic[T]):
"""
A value that indicates success and which stores arbitrary data for the return value.
"""
__match_args__ = ("ok_value",)
__slots__ = ("_value",)
def __iter__(self) -> Iterator[T]:
yield self._value
def __init__(self, value: T) -> None:
self._value = value
def __repr__(self) -> str:
return "Ok({})".format(repr(self._value))
def __eq__(self, other: Any) -> bool:
return isinstance(other, Ok) and self._value == other._value
def __ne__(self, other: Any) -> bool:
return not (self == other)
def __hash__(self) -> int:
return hash((True, self._value))
def is_ok(self) -> Literal[True]:
return True
def is_err(self) -> Literal[False]:
return False
def ok(self) -> T:
"""
Return the value.
"""
return self._value
def err(self) -> None:
"""
Return `None`.
"""
return None
@property
def value(self) -> T:
"""
Return the inner value.
@deprecated Use `ok_value` or `err_value` instead. This method will be
removed in a future version.
"""
warn(
"Accessing `.value` on Result type is deprecated, please use "
+ "`.ok_value` or `.err_value` instead",
DeprecationWarning,
stacklevel=2,
)
return self._value
@property
def ok_value(self) -> T:
"""
Return the inner value.
"""
return self._value
def expect(self, _message: str) -> T:
"""
Return the value.
"""
return self._value
def expect_err(self, message: str) -> NoReturn:
"""
Raise an UnwrapError since this type is `Ok`
"""
raise UnwrapError(self, message)
def unwrap(self) -> T:
"""
Return the value.
"""
return self._value
def unwrap_err(self) -> NoReturn:
"""
Raise an UnwrapError since this type is `Ok`
"""
raise UnwrapError(self, "Called `Result.unwrap_err()` on an `Ok` value")
def unwrap_or(self, _default: U) -> T:
"""
Return the value.
"""
return self._value
def unwrap_or_else(self, op: object) -> T:
"""
Return the value.
"""
return self._value
def unwrap_or_raise(self, e: object) -> T:
"""
Return the value.
"""
return self._value
def map(self, op: Callable[[T], U]) -> Ok[U]:
"""
The contained result is `Ok`, so return `Ok` with original value mapped to
a new value using the passed in function.
"""
return Ok(op(self._value))
async def map_async(
self, op: Callable[[T], Awaitable[U]]
) -> Ok[U]:
"""
The contained result is `Ok`, so return the result of `op` with the
original value passed in
"""
return Ok(await op(self._value))
def map_or(self, default: object, op: Callable[[T], U]) -> U:
"""
The contained result is `Ok`, so return the original value mapped to a new
value using the passed in function.
"""
return op(self._value)
def map_or_else(self, default_op: object, op: Callable[[T], U]) -> U:
"""
The contained result is `Ok`, so return original value mapped to
a new value using the passed in `op` function.
"""
return op(self._value)
def map_err(self, op: object) -> Ok[T]:
"""
The contained result is `Ok`, so return `Ok` with the original value
"""
return self
def and_then(self, op: Callable[[T], Result[U, E]]) -> Result[U, E]:
"""
The contained result is `Ok`, so return the result of `op` with the
original value passed in
"""
return op(self._value)
async def and_then_async(
self, op: Callable[[T], Awaitable[Result[U, E]]]
) -> Result[U, E]:
"""
The contained result is `Ok`, so return the result of `op` with the
original value passed in
"""
return await op(self._value)
def or_else(self, op: object) -> Ok[T]:
"""
The contained result is `Ok`, so return `Ok` with the original value
"""
return self
def inspect(self, op: Callable[[T], Any]) -> Result[T, E]:
"""
Calls a function with the contained value if `Ok`. Returns the original result.
"""
op(self._value)
return self
def inspect_err(self, op: Callable[[E], Any]) -> Result[T, E]:
"""
Calls a function with the contained value if `Err`. Returns the original result.
"""
return self
class DoException(Exception):
"""
This is used to signal to `do()` that the result is an `Err`,
which short-circuits the generator and returns that Err.
Using this exception for control flow in `do()` allows us
to simulate `and_then()` in the Err case: namely, we don't call `op`,
we just return `self` (the Err).
"""
def __init__(self, err: Err[E]) -> None:
self.err = err
class Err(Generic[E]):
"""
A value that signifies failure and which stores arbitrary data for the error.
"""
__match_args__ = ("err_value",)
__slots__ = ("_value",)
def __iter__(self) -> Iterator[NoReturn]:
def _iter() -> Iterator[NoReturn]:
# Exception will be raised when the iterator is advanced, not when it's created
raise DoException(self)
yield # This yield will never be reached, but is necessary to create a generator
return _iter()
def __init__(self, value: E) -> None:
self._value = value
def __repr__(self) -> str:
return "Err({})".format(repr(self._value))
def __eq__(self, other: Any) -> bool:
return isinstance(other, Err) and self._value == other._value
def __ne__(self, other: Any) -> bool:
return not (self == other)
def __hash__(self) -> int:
return hash((False, self._value))
def is_ok(self) -> Literal[False]:
return False
def is_err(self) -> Literal[True]:
return True
def ok(self) -> None:
"""
Return `None`.
"""
return None
def err(self) -> E:
"""
Return the error.
"""
return self._value
@property
def value(self) -> E:
"""
Return the inner value.
@deprecated Use `ok_value` or `err_value` instead. This method will be
removed in a future version.
"""
warn(
"Accessing `.value` on Result type is deprecated, please use "
+ "`.ok_value` or '.err_value' instead",
DeprecationWarning,
stacklevel=2,
)
return self._value
@property
def err_value(self) -> E:
"""
Return the inner value.
"""
return self._value
def expect(self, message: str) -> NoReturn:
"""
Raises an `UnwrapError`.
"""
exc = UnwrapError(
self,
f"{message}: {self._value!r}",
)
if isinstance(self._value, BaseException):
raise exc from self._value
raise exc
def expect_err(self, _message: str) -> E:
"""
Return the inner value
"""
return self._value
def unwrap(self) -> NoReturn:
"""
Raises an `UnwrapError`.
"""
exc = UnwrapError(
self,
f"Called `Result.unwrap()` on an `Err` value: {self._value!r}",
)
if isinstance(self._value, BaseException):
raise exc from self._value
raise exc
def unwrap_err(self) -> E:
"""
Return the inner value
"""
return self._value
def unwrap_or(self, default: U) -> U:
"""
Return `default`.
"""
return default
def unwrap_or_else(self, op: Callable[[E], T]) -> T:
"""
The contained result is ``Err``, so return the result of applying
``op`` to the error value.
"""
return op(self._value)
def unwrap_or_raise(self, e: Type[TBE]) -> NoReturn:
"""
The contained result is ``Err``, so raise the exception with the value.
"""
raise e(self._value)
def map(self, op: object) -> Err[E]:
"""
Return `Err` with the same value
"""
return self
async def map_async(self, op: object) -> Err[E]:
"""
The contained result is `Ok`, so return the result of `op` with the
original value passed in
"""
return self
def map_or(self, default: U, op: object) -> U:
"""
Return the default value
"""
return default
def map_or_else(self, default_op: Callable[[], U], op: object) -> U:
"""
Return the result of the default operation
"""
return default_op()
def map_err(self, op: Callable[[E], F]) -> Err[F]:
"""
The contained result is `Err`, so return `Err` with original error mapped to
a new value using the passed in function.
"""
return Err(op(self._value))
def and_then(self, op: object) -> Err[E]:
"""
The contained result is `Err`, so return `Err` with the original value
"""
return self
async def and_then_async(self, op: object) -> Err[E]:
"""
The contained result is `Err`, so return `Err` with the original value
"""
return self
def or_else(self, op: Callable[[E], Result[T, F]]) -> Result[T, F]:
"""
The contained result is `Err`, so return the result of `op` with the
original value passed in
"""
return op(self._value)
def inspect(self, op: Callable[[T], Any]) -> Result[T, E]:
"""
Calls a function with the contained value if `Ok`. Returns the original result.
"""
return self
def inspect_err(self, op: Callable[[E], Any]) -> Result[T, E]:
"""
Calls a function with the contained value if `Err`. Returns the original result.
"""
op(self._value)
return self
# define Result as a generic type alias for use
# in type annotations
"""
A simple `Result` type inspired by Rust.
Not all methods (https://doc.rust-lang.org/std/result/enum.Result.html)
have been implemented, only the ones that make sense in the Python context.
"""
Result: TypeAlias = Union[Ok[T], Err[E]]
"""
A type to use in `isinstance` checks.
This is purely for convenience sake, as you could also just write `isinstance(res, (Ok, Err))
"""
OkErr: Final = (Ok, Err)
class UnwrapError(Exception):
"""
Exception raised from ``.unwrap_<...>`` and ``.expect_<...>`` calls.
The original ``Result`` can be accessed via the ``.result`` attribute, but
this is not intended for regular use, as type information is lost:
``UnwrapError`` doesn't know about both ``T`` and ``E``, since it's raised
from ``Ok()`` or ``Err()`` which only knows about either ``T`` or ``E``,
not both.
"""
_result: Result[object, object]
def __init__(self, result: Result[object, object], message: str) -> None:
self._result = result
super().__init__(message)
@property
def result(self) -> Result[Any, Any]:
"""
Returns the original result.
"""
return self._result
def as_result(
*exceptions: Type[TBE],
) -> Callable[[Callable[P, R]], Callable[P, Result[R, TBE]]]:
"""
Make a decorator to turn a function into one that returns a ``Result``.
Regular return values are turned into ``Ok(return_value)``. Raised
exceptions of the specified exception type(s) are turned into ``Err(exc)``.
"""
if not exceptions or not all(
inspect.isclass(exception) and issubclass(exception, BaseException)
for exception in exceptions
):
raise TypeError("as_result() requires one or more exception types")
def decorator(f: Callable[P, R]) -> Callable[P, Result[R, TBE]]:
"""
Decorator to turn a function into one that returns a ``Result``.
"""
@functools.wraps(f)
def wrapper(*args: P.args, **kwargs: P.kwargs) -> Result[R, TBE]:
try:
return Ok(f(*args, **kwargs))
except exceptions as exc:
return Err(exc)
return wrapper
return decorator
def as_async_result(
*exceptions: Type[TBE],
) -> Callable[[Callable[P, Awaitable[R]]], Callable[P, Awaitable[Result[R, TBE]]]]:
"""
Make a decorator to turn an async function into one that returns a ``Result``.
Regular return values are turned into ``Ok(return_value)``. Raised
exceptions of the specified exception type(s) are turned into ``Err(exc)``.
"""
if not exceptions or not all(
inspect.isclass(exception) and issubclass(exception, BaseException)
for exception in exceptions
):
raise TypeError("as_result() requires one or more exception types")
def decorator(
f: Callable[P, Awaitable[R]]
) -> Callable[P, Awaitable[Result[R, TBE]]]:
"""
Decorator to turn a function into one that returns a ``Result``.
"""
@functools.wraps(f)
async def async_wrapper(*args: P.args, **kwargs: P.kwargs) -> Result[R, TBE]:
try:
return Ok(await f(*args, **kwargs))
except exceptions as exc:
return Err(exc)
return async_wrapper
return decorator
def is_ok(result: Result[T, E]) -> TypeIs[Ok[T]]:
"""A type guard to check if a result is an Ok
Usage:
``` python
r: Result[int, str] = get_a_result()
if is_ok(r):
r # r is of type Ok[int]
elif is_err(r):
r # r is of type Err[str]
```
"""
return result.is_ok()
def is_err(result: Result[T, E]) -> TypeIs[Err[E]]:
"""A type guard to check if a result is an Err
Usage:
``` python
r: Result[int, str] = get_a_result()
if is_ok(r):
r # r is of type Ok[int]
elif is_err(r):
r # r is of type Err[str]
```
"""
return result.is_err()
def do(gen: Generator[Result[T, E], None, None]) -> Result[T, E]:
"""Do notation for Result (syntactic sugar for sequence of `and_then()` calls).
Usage:
``` rust
// This is similar to
use do_notation::m;
let final_result = m! {
x <- Ok("hello");
y <- Ok(True);
Ok(len(x) + int(y) + 0.5)
};
```
``` rust
final_result: Result[float, int] = do(
Ok(len(x) + int(y) + 0.5)
for x in Ok("hello")
for y in Ok(True)
)
```
NOTE: If you exclude the type annotation e.g. `Result[float, int]`
your type checker might be unable to infer the return type.
To avoid an error, you might need to help it with the type hint.
"""
try:
return next(gen)
except DoException as e:
out: Err[E] = e.err # type: ignore
return out
except TypeError as te:
# Turn this into a more helpful error message.
# Python has strange rules involving turning generators involving `await`
# into async generators, so we want to make sure to help the user clearly.
if "'async_generator' object is not an iterator" in str(te):
raise TypeError(
"Got async_generator but expected generator."
"See the section on do notation in the README."
)
raise te
async def do_async(
gen: Union[Generator[Result[T, E], None, None], AsyncGenerator[Result[T, E], None]]
) -> Result[T, E]:
"""Async version of do. Example:
``` python
final_result: Result[float, int] = await do_async(
Ok(len(x) + int(y) + z)
for x in await get_async_result_1()
for y in await get_async_result_2()
for z in get_sync_result_3()
)
```
NOTE: Python makes generators async in a counter-intuitive way.
``` python
# This is a regular generator:
async def foo(): ...
do(Ok(1) for x in await foo())
```
``` python
# But this is an async generator:
async def foo(): ...
async def bar(): ...
do(
Ok(1)
for x in await foo()
for y in await bar()
)
```
We let users try to use regular `do()`, which works in some cases
of awaiting async values. If we hit a case like above, we raise
an exception telling the user to use `do_async()` instead.
See `do()`.
However, for better usability, it's better for `do_async()` to also accept
regular generators, as you get in the first case:
``` python
async def foo(): ...
do(Ok(1) for x in await foo())
```
Furthermore, neither mypy nor pyright can infer that the second case is
actually an async generator, so we cannot annotate `do_async()`
as accepting only an async generator. This is additional motivation
to accept either.
"""
try:
if isinstance(gen, AsyncGenerator):
return await gen.__anext__()
else:
return next(gen)
except DoException as e:
out: Err[E] = e.err # type: ignore
return out

View file

@ -0,0 +1,29 @@
from __future__ import annotations
from result import Err, Ok, Result
def test_pattern_matching_on_ok_type() -> None:
"""
Pattern matching on ``Ok()`` matches the contained value.
"""
o: Result[str, int] = Ok("yay")
match o:
case Ok(value):
reached = True
assert value == "yay"
assert reached
def test_pattern_matching_on_err_type() -> None:
"""
Pattern matching on ``Err()`` matches the contained value.
"""
n: Result[int, str] = Err("nay")
match n:
case Err(value):
reached = True
assert value == "nay"
assert reached

View file

@ -0,0 +1,430 @@
from __future__ import annotations
from typing import Callable
import pytest
from result import Err, Ok, OkErr, Result, UnwrapError, as_async_result, as_result
def test_ok_factories() -> None:
instance = Ok(1)
assert instance._value == 1
assert instance.is_ok() is True
def test_err_factories() -> None:
instance = Err(2)
assert instance._value == 2
assert instance.is_err() is True
def test_eq() -> None:
assert Ok(1) == Ok(1)
assert Err(1) == Err(1)
assert Ok(1) != Err(1)
assert Ok(1) != Ok(2)
assert Err(1) != Err(2)
assert not (Ok(1) != Ok(1))
assert Ok(1) != "abc"
assert Ok("0") != Ok(0)
def test_hash() -> None:
assert len({Ok(1), Err("2"), Ok(1), Err("2")}) == 2
assert len({Ok(1), Ok(2)}) == 2
assert len({Ok("a"), Err("a")}) == 2
def test_repr() -> None:
"""
``repr()`` returns valid code if the wrapped value's ``repr()`` does as well.
"""
o = Ok(123)
n = Err(-1)
assert repr(o) == "Ok(123)"
assert o == eval(repr(o))
assert repr(n) == "Err(-1)"
assert n == eval(repr(n))
def test_ok_value() -> None:
res = Ok("haha")
assert res.ok_value == "haha"
def test_err_value() -> None:
res = Err("haha")
assert res.err_value == "haha"
def test_ok() -> None:
res = Ok("haha")
assert res.is_ok() is True
assert res.is_err() is False
assert res.ok_value == "haha"
def test_err() -> None:
res = Err(":(")
assert res.is_ok() is False
assert res.is_err() is True
assert res.err_value == ":("
def test_err_value_is_exception() -> None:
res = Err(ValueError("Some Error"))
assert res.is_ok() is False
assert res.is_err() is True
with pytest.raises(UnwrapError):
res.unwrap()
try:
res.unwrap()
except UnwrapError as e:
cause = e.__cause__
assert isinstance(cause, ValueError)
def test_ok_method() -> None:
o = Ok("yay")
n = Err("nay")
assert o.ok() == "yay"
assert n.ok() is None # type: ignore[func-returns-value]
def test_err_method() -> None:
o = Ok("yay")
n = Err("nay")
assert o.err() is None # type: ignore[func-returns-value]
assert n.err() == "nay"
def test_expect() -> None:
o = Ok("yay")
n = Err("nay")
assert o.expect("failure") == "yay"
with pytest.raises(UnwrapError):
n.expect("failure")
def test_expect_err() -> None:
o = Ok("yay")
n = Err("nay")
assert n.expect_err("hello") == "nay"
with pytest.raises(UnwrapError):
o.expect_err("hello")
def test_unwrap() -> None:
o = Ok("yay")
n = Err("nay")
assert o.unwrap() == "yay"
with pytest.raises(UnwrapError):
n.unwrap()
def test_unwrap_err() -> None:
o = Ok("yay")
n = Err("nay")
assert n.unwrap_err() == "nay"
with pytest.raises(UnwrapError):
o.unwrap_err()
def test_unwrap_or() -> None:
o = Ok("yay")
n = Err("nay")
assert o.unwrap_or("some_default") == "yay"
assert n.unwrap_or("another_default") == "another_default"
def test_unwrap_or_else() -> None:
o = Ok("yay")
n = Err("nay")
assert o.unwrap_or_else(str.upper) == "yay"
assert n.unwrap_or_else(str.upper) == "NAY"
def test_unwrap_or_raise() -> None:
o = Ok("yay")
n = Err("nay")
assert o.unwrap_or_raise(ValueError) == "yay"
with pytest.raises(ValueError) as exc_info:
n.unwrap_or_raise(ValueError)
assert exc_info.value.args == ("nay",)
def test_map() -> None:
o = Ok("yay")
n = Err("nay")
assert o.map(str.upper).ok() == "YAY"
assert n.map(str.upper).err() == "nay"
num = Ok(3)
errnum = Err(2)
assert num.map(str).ok() == "3"
assert errnum.map(str).err() == 2
def test_map_or() -> None:
o = Ok("yay")
n = Err("nay")
assert o.map_or("hay", str.upper) == "YAY"
assert n.map_or("hay", str.upper) == "hay"
num = Ok(3)
errnum = Err(2)
assert num.map_or("-1", str) == "3"
assert errnum.map_or("-1", str) == "-1"
def test_map_or_else() -> None:
o = Ok("yay")
n = Err("nay")
assert o.map_or_else(lambda: "hay", str.upper) == "YAY"
assert n.map_or_else(lambda: "hay", str.upper) == "hay"
num = Ok(3)
errnum = Err(2)
assert num.map_or_else(lambda: "-1", str) == "3"
assert errnum.map_or_else(lambda: "-1", str) == "-1"
def test_map_err() -> None:
o = Ok("yay")
n = Err("nay")
assert o.map_err(str.upper).ok() == "yay"
assert n.map_err(str.upper).err() == "NAY"
def test_and_then() -> None:
assert Ok(2).and_then(sq).and_then(sq).ok() == 16
assert Ok(2).and_then(sq).and_then(to_err).err() == 4
assert Ok(2).and_then(to_err).and_then(sq).err() == 2
assert Err(3).and_then(sq).and_then(sq).err() == 3
assert Ok(2).and_then(sq_lambda).and_then(sq_lambda).ok() == 16
assert Ok(2).and_then(sq_lambda).and_then(to_err_lambda).err() == 4
assert Ok(2).and_then(to_err_lambda).and_then(sq_lambda).err() == 2
assert Err(3).and_then(sq_lambda).and_then(sq_lambda).err() == 3
def test_inspect() -> None:
oks: list[int] = []
add_to_oks: Callable[[int], None] = lambda x: oks.append(x)
assert Ok(2).inspect(add_to_oks) == Ok(2)
assert Err("e").inspect(add_to_oks) == Err("e")
assert oks == [2]
def test_inspect_err() -> None:
errs: list[str] = []
add_to_errs: Callable[[str], None] = lambda x: errs.append(x)
assert Ok(2).inspect_err(add_to_errs) == Ok(2)
assert Err("e").inspect_err(add_to_errs) == Err("e")
assert errs == ["e"]
def test_inspect_regular_fn() -> None:
oks: list[str] = []
def _add_to_oks(x: str) -> str:
oks.append(x)
return x + x
assert Ok("hello").inspect(_add_to_oks) == Ok("hello")
assert Err("error").inspect(_add_to_oks) == Err("error")
assert oks == ["hello"]
@pytest.mark.asyncio
async def test_and_then_async() -> None:
assert (
await (await Ok(2).and_then_async(sq_async)).and_then_async(sq_async)
).ok() == 16
assert (
await (await Ok(2).and_then_async(sq_async)).and_then_async(to_err_async)
).err() == 4
assert (
await (await Ok(2).and_then_async(to_err_async)).and_then_async(to_err_async)
).err() == 2
assert (
await (await Err(3).and_then_async(sq_async)).and_then_async(sq_async)
).err() == 3
@pytest.mark.asyncio
async def test_map_async() -> None:
async def str_upper_async(s: str) -> str:
return s.upper()
async def str_async(x: int) -> str:
return str(x)
o = Ok("yay")
n = Err("nay")
assert (await o.map_async(str_upper_async)).ok() == "YAY"
assert (await n.map_async(str_upper_async)).err() == "nay"
num = Ok(3)
errnum = Err(2)
assert (await num.map_async(str_async)).ok() == "3"
assert (await errnum.map_async(str_async)).err() == 2
def test_or_else() -> None:
assert Ok(2).or_else(sq).or_else(sq).ok() == 2
assert Ok(2).or_else(to_err).or_else(sq).ok() == 2
assert Err(3).or_else(sq).or_else(to_err).ok() == 9
assert Err(3).or_else(to_err).or_else(to_err).err() == 3
assert Ok(2).or_else(sq_lambda).or_else(sq).ok() == 2
assert Ok(2).or_else(to_err_lambda).or_else(sq_lambda).ok() == 2
assert Err(3).or_else(sq_lambda).or_else(to_err_lambda).ok() == 9
assert Err(3).or_else(to_err_lambda).or_else(to_err_lambda).err() == 3
def test_isinstance_result_type() -> None:
o = Ok("yay")
n = Err("nay")
assert isinstance(o, OkErr)
assert isinstance(n, OkErr)
assert not isinstance(1, OkErr)
def test_error_context() -> None:
n = Err("nay")
with pytest.raises(UnwrapError) as exc_info:
n.unwrap()
exc = exc_info.value
assert exc.result is n
def test_slots() -> None:
"""
Ok and Err have slots, so assigning arbitrary attributes fails.
"""
o = Ok("yay")
n = Err("nay")
with pytest.raises(AttributeError):
o.some_arbitrary_attribute = 1 # type: ignore[attr-defined]
with pytest.raises(AttributeError):
n.some_arbitrary_attribute = 1 # type: ignore[attr-defined]
def test_as_result() -> None:
"""
``as_result()`` turns functions into ones that return a ``Result``.
"""
@as_result(ValueError)
def good(value: int) -> int:
return value
@as_result(IndexError, ValueError)
def bad(value: int) -> int:
raise ValueError
good_result = good(123)
bad_result = bad(123)
assert isinstance(good_result, Ok)
assert good_result.unwrap() == 123
assert isinstance(bad_result, Err)
assert isinstance(bad_result.unwrap_err(), ValueError)
def test_as_result_other_exception() -> None:
"""
``as_result()`` only catches the specified exceptions.
"""
@as_result(ValueError)
def f() -> int:
raise IndexError
with pytest.raises(IndexError):
f()
def test_as_result_invalid_usage() -> None:
"""
Invalid use of ``as_result()`` raises reasonable errors.
"""
message = "requires one or more exception types"
with pytest.raises(TypeError, match=message):
@as_result() # No exception types specified
def f() -> int:
return 1
with pytest.raises(TypeError, match=message):
@as_result("not an exception type") # type: ignore[arg-type]
def g() -> int:
return 1
def test_as_result_type_checking() -> None:
"""
The ``as_result()`` is a signature-preserving decorator.
"""
@as_result(ValueError)
def f(a: int) -> int:
return a
res: Result[int, ValueError]
res = f(123) # No mypy error here.
assert res.ok() == 123
@pytest.mark.asyncio
async def test_as_async_result() -> None:
"""
``as_async_result()`` turns functions into ones that return a ``Result``.
"""
@as_async_result(ValueError)
async def good(value: int) -> int:
return value
@as_async_result(IndexError, ValueError)
async def bad(value: int) -> int:
raise ValueError
good_result = await good(123)
bad_result = await bad(123)
assert isinstance(good_result, Ok)
assert good_result.unwrap() == 123
assert isinstance(bad_result, Err)
assert isinstance(bad_result.unwrap_err(), ValueError)
def sq(i: int) -> Result[int, int]:
return Ok(i * i)
async def sq_async(i: int) -> Result[int, int]:
return Ok(i * i)
def to_err(i: int) -> Result[int, int]:
return Err(i)
async def to_err_async(i: int) -> Result[int, int]:
return Err(i)
# Lambda versions of the same functions, just for test/type coverage
sq_lambda: Callable[[int], Result[int, int]] = lambda i: Ok(i * i)
to_err_lambda: Callable[[int], Result[int, int]] = lambda i: Err(i)

View file

@ -0,0 +1,227 @@
from __future__ import annotations
import pytest
from result import Err, Ok, Result, do, do_async
def test_result_do_general() -> None:
def resx(is_suc: bool) -> Result[str, int]:
return Ok("hello") if is_suc else Err(1)
def resy(is_suc: bool) -> Result[bool, int]:
return Ok(True) if is_suc else Err(2)
def _get_output(is_suc1: bool, is_suc2: bool) -> Result[float, int]:
out: Result[float, int] = do(
Ok(len(x) + int(y) + 0.5) for x in resx(is_suc1) for y in resy(is_suc2)
)
return out
assert _get_output(True, True) == Ok(6.5)
assert _get_output(True, False) == Err(2)
assert _get_output(False, True) == Err(1)
assert _get_output(False, False) == Err(1)
def _get_output_return_immediately(
is_suc1: bool, is_suc2: bool
) -> Result[float, int]:
return do(
Ok(len(x) + int(y) + 0.5) for x in resx(is_suc1) for y in resy(is_suc2)
)
assert _get_output_return_immediately(True, True) == Ok(6.5)
@pytest.mark.asyncio
async def test_result_do_general_with_async_values() -> None:
# Asyncio works with regular `do()` as long as you await
# the async calls outside the `do()` expression.
# This causes the generator to be a regular (not async) generator.
async def aget_resx(is_suc: bool) -> Result[str, int]:
return Ok("hello") if is_suc else Err(1)
async def aget_resy(is_suc: bool) -> Result[bool, int]:
return Ok(True) if is_suc else Err(2)
async def _aget_output(is_suc1: bool, is_suc2: bool) -> Result[float, int]:
resx, resy = await aget_resx(is_suc1), await aget_resy(is_suc2)
out: Result[float, int] = do(
Ok(len(x) + int(y) + 0.5) for x in resx for y in resy
)
return out
assert await _aget_output(True, True) == Ok(6.5)
assert await _aget_output(True, False) == Err(2)
assert await _aget_output(False, True) == Err(1)
assert await _aget_output(False, False) == Err(1)
@pytest.mark.asyncio
async def test_result_do_async_one_value() -> None:
"""This is a strange case where Python creates a regular
(non async) generator despite an `await` inside the generator expression.
For convenience, although this works with regular `do()`, we want to support this
with `do_async()` as well."""
async def aget_resx(is_suc: bool) -> Result[str, int]:
return Ok("hello") if is_suc else Err(1)
def get_resz(is_suc: bool) -> Result[float, int]:
return Ok(0.5) if is_suc else Err(3)
assert await do_async(Ok(len(x)) for x in await aget_resx(True)) == Ok(5)
assert await do_async(Ok(len(x)) for x in await aget_resx(False)) == Err(1)
async def _aget_output(is_suc1: bool, is_suc3: bool) -> Result[float, int]:
return await do_async(
Ok(len(x) + z) for x in await aget_resx(is_suc1) for z in get_resz(is_suc3)
)
assert await _aget_output(True, True) == Ok(5.5)
assert await _aget_output(True, False) == Err(3)
assert await _aget_output(False, True) == Err(1)
assert await _aget_output(False, False) == Err(1)
@pytest.mark.asyncio
async def test_result_do_async_general() -> None:
async def aget_resx(is_suc: bool) -> Result[str, int]:
return Ok("hello") if is_suc else Err(1)
async def aget_resy(is_suc: bool) -> Result[bool, int]:
return Ok(True) if is_suc else Err(2)
def get_resz(is_suc: bool) -> Result[float, int]:
return Ok(0.5) if is_suc else Err(3)
async def _aget_output(
is_suc1: bool, is_suc2: bool, is_suc3: bool
) -> Result[float, int]:
out: Result[float, int] = await do_async(
Ok(len(x) + int(y) + z)
for x in await aget_resx(is_suc1)
for y in await aget_resy(is_suc2)
for z in get_resz(is_suc3)
)
return out
assert await _aget_output(True, True, True) == Ok(6.5)
assert await _aget_output(True, False, True) == Err(2)
assert await _aget_output(False, True, True) == Err(1)
assert await _aget_output(False, False, True) == Err(1)
assert await _aget_output(True, True, False) == Err(3)
assert await _aget_output(True, False, False) == Err(2)
assert await _aget_output(False, True, False) == Err(1)
assert await _aget_output(False, False, False) == Err(1)
async def _aget_output_return_immediately(
is_suc1: bool, is_suc2: bool, is_suc3: bool
) -> Result[float, int]:
return await do_async(
Ok(len(x) + int(y) + z)
for x in await aget_resx(is_suc1)
for y in await aget_resy(is_suc2)
for z in get_resz(is_suc3)
)
assert await _aget_output_return_immediately(True, True, True) == Ok(6.5)
@pytest.mark.asyncio
async def test_result_do_async_further_processing() -> None:
async def aget_resx(is_suc: bool) -> Result[str, int]:
return Ok("hello") if is_suc else Err(1)
async def aget_resy(is_suc: bool) -> Result[bool, int]:
return Ok(True) if is_suc else Err(2)
def get_resz(is_suc: bool) -> Result[float, int]:
return Ok(0.5) if is_suc else Err(3)
async def process_xyz(x: str, y: bool, z: float) -> Result[float, int]:
return Ok(len(x) + int(y) + z)
async def _aget_output(
is_suc1: bool, is_suc2: bool, is_suc3: bool
) -> Result[float, int]:
out: Result[float, int] = await do_async(
Ok(w)
for x in await aget_resx(is_suc1)
for y in await aget_resy(is_suc2)
for z in get_resz(is_suc3)
for w in await process_xyz(x, y, z)
)
return out
assert await _aget_output(True, True, True) == Ok(6.5)
assert await _aget_output(True, False, True) == Err(2)
assert await _aget_output(False, True, True) == Err(1)
assert await _aget_output(False, False, True) == Err(1)
assert await _aget_output(True, True, False) == Err(3)
assert await _aget_output(True, False, False) == Err(2)
assert await _aget_output(False, True, False) == Err(1)
assert await _aget_output(False, False, False) == Err(1)
@pytest.mark.asyncio
async def test_result_do_general_with_async_values_inline_error() -> None:
"""
Due to subtle behavior, `do()` works in certain cases involving async
calls but not others. We surface a more helpful error to the user
in cases where it doesn't work indicating to use `do_async()` instead.
Contrast this with `test_result_do_general_with_async_values()`
in which using `do()` works with async functions as long as
their return values are resolved outside the `do()` expression.
"""
async def aget_resx(is_suc: bool) -> Result[str, int]:
return Ok("hello") if is_suc else Err(1)
async def aget_resy(is_suc: bool) -> Result[bool, int]:
return Ok(True) if is_suc else Err(2)
def get_resz(is_suc: bool) -> Result[float, int]:
return Ok(0.5) if is_suc else Err(3)
with pytest.raises(TypeError) as excinfo:
do(
Ok(len(x) + int(y) + z)
for x in await aget_resx(True)
for y in await aget_resy(True)
for z in get_resz(True)
)
assert (
"Got async_generator but expected generator.See the section on do notation in the README."
) in excinfo.value.args[0]
@pytest.mark.asyncio
async def test_result_do_async_swap_order() -> None:
def foo() -> Result[int, str]:
return Ok(1)
async def bar() -> Result[int, str]:
return Ok(2)
result1: Result[int, str] = await do_async(
Ok(x + y)
# x first
for x in foo()
# then y
for y in await bar()
)
result2: Result[int, str] = await do_async(
Ok(x + y)
# y first
for y in await bar()
# then x
for x in foo()
)
assert result1 == result2 == Ok(3)

View file

@ -0,0 +1,100 @@
---
# reveal_type(res3) # N: Revealed type is "result.result.Err[builtins.int]"
- case: failure_lash
disable_cache: false
main: |
from typing import Callable, List, Optional
from result import Result, Ok, Err
res1: Result[str, int] = Ok('hello')
reveal_type(res1) # N: Revealed type is "Union[result.result.Ok[builtins.str], result.result.Err[builtins.int]]"
if isinstance(res1, Ok):
ok: Ok[str] = res1
reveal_type(ok) # N: Revealed type is "result.result.Ok[builtins.str]"
okValue: str = res1.ok()
reveal_type(okValue) # N: Revealed type is "builtins.str"
mapped_to_float: float = res1.map_or(1.0, lambda s: len(s) * 1.5)
reveal_type(mapped_to_float) # N: Revealed type is "builtins.float"
else:
err: Err[int] = res1
reveal_type(err) # N: Revealed type is "result.result.Err[builtins.int]"
errValue: int = err.err()
reveal_type(errValue) # N: Revealed type is "builtins.int"
mapped_to_list: Optional[List[int]] = res1.map_err(lambda e: [e]).err()
reveal_type(mapped_to_list) # N: Revealed type is "Union[builtins.list[builtins.int], None]"
# Test constructor functions
res2 = Ok(42)
reveal_type(res2) # N: Revealed type is "result.result.Ok[builtins.int]"
res3 = Err(1)
reveal_type(res3) # N: Revealed type is "result.result.Err[builtins.int]"
res4 = Ok(4)
add1: Callable[[int], Result[int, str]] = lambda i: Ok(i + 1)
toint: Callable[[str], Result[int, str]] = lambda i: Ok(int(i))
res5 = res4.and_then(add1)
reveal_type(res5) # N: Revealed type is "Union[result.result.Ok[builtins.int], result.result.Err[builtins.str]]"
res6 = res4.or_else(toint)
reveal_type(res6) # N: Revealed type is "result.result.Ok[builtins.int]"
- case: covariance
disable_cache: false
main: |
from result import Result, Ok, Err
ok_int: Ok[int] = Ok(42)
ok_float: Ok[float] = ok_int
ok_int = ok_float # E: Incompatible types in assignment (expression has type "Ok[float]", variable has type "Ok[int]") [assignment]
err_type: Err[TypeError] = Err(TypeError("foo"))
err_exc: Err[Exception] = err_type
err_type = err_exc # E: Incompatible types in assignment (expression has type "Err[Exception]", variable has type "Err[TypeError]") [assignment]
result_int_type: Result[int, TypeError] = ok_int or err_type
result_float_exc: Result[float, Exception] = result_int_type
result_int_type = result_float_exc # E: Incompatible types in assignment (expression has type "Ok[float] | Err[Exception]", variable has type "Ok[int] | Err[TypeError]") [assignment]
- case: map_ok_err
disable_cache: false
main: |
from result import Err, Ok
o = Ok("42")
reveal_type(o.map(int)) # N: Revealed type is "result.result.Ok[builtins.int]"
reveal_type(o.map_err(int)) # N: Revealed type is "result.result.Ok[builtins.str]"
e = Err("42")
reveal_type(e.map(int)) # N: Revealed type is "result.result.Err[builtins.str]"
reveal_type(e.map_err(int)) # N: Revealed type is "result.result.Err[builtins.int]"
- case: map_result
disable_cache: false
main: |
from result import Result, Ok
greeting_res: Result[str, ValueError] = Ok("Hello")
personalized_greeting_res = greeting_res.map(lambda g: f"{g}, John")
reveal_type(personalized_greeting_res) # N: Revealed type is "Union[result.result.Ok[builtins.str], result.result.Err[builtins.ValueError]]"
personalized_greeting = personalized_greeting_res.ok()
reveal_type(personalized_greeting) # N: Revealed type is "Union[builtins.str, None]"
- case: map_result
disable_cache: false
main: |
from result import Result, Ok, Err, is_ok, is_err
res1: Result[int, str] = Ok(1)
if is_ok(res1):
reveal_type(res1) # N: Revealed type is "result.result.Ok[builtins.int]"
else:
reveal_type(res1) # N: Revealed type is "result.result.Err[builtins.str]"
res2: Result[int, str] = Err("error")
if is_err(res2):
reveal_type(res2) # N: Revealed type is "result.result.Err[builtins.str]"
else:
reveal_type(res2) # N: Revealed type is "result.result.Ok[builtins.int]"