Result
is obviously a result of some series of computations.
It might succeed with some resulting value.
Or it might return an error with some extra details.
Result
consist of two types: Success
and Failure
.
Success
represents successful operation result
and Failure
indicates that something has failed.
from returns.result import Result, Success, Failure
def find_user(user_id: int) -> Result['User', str]:
user = User.objects.filter(id=user_id)
if user.exists():
return Success(user[0])
return Failure('User was not found')
user_search_result = find_user(1)
# => Success(User{id: 1, ...})
user_search_result = find_user(0) # id 0 does not exist!
# => Failure('User was not found')
When is it useful?
When you do not want to use exceptions to break your execution scope.
Or when you do not want to use None
to represent empty values,
since it will raise TypeError
somewhere
and other None
exception-friends.
is_succesful
is used to
tell whether or not your result is a success.
We treat only treat types that does not throw as a successful ones,
basically: Success
.
from returns.result import Success, Failure, is_successful
is_successful(Success(1))
# => True
is_successful(Failure('text'))
# => False
What is a pipeline
?
It is a more user-friendly syntax to work with containers
that support both async and regular functions.
Consider this task. We were asked to create a method that will connect together a simple pipeline of three steps:
We validate passed username
and email
We create a new Account
with this data, if it does not exists
We create a new User
associated with the Account
And we know that this pipeline can fail in several places:
Wrong username
or email
might be passed, so the validation will fail
Account
with this username
or email
might already exist
User
creation might fail as well,
since it also makes an HTTP
request to another micro-service deep inside
Here’s the code to illustrate the task.
from returns.result import Result, Success, Failure, pipeline
class CreateAccountAndUser(object):
"""Creates new Account-User pair."""
# TODO: we need to create a pipeline of these methods somehow...
# Protected methods
def _validate_user(
self, username: str, email: str,
) -> Result['UserSchema', str]:
"""Returns an UserSchema for valid input, otherwise a Failure."""
def _create_account(
self, user_schema: 'UserSchema',
) -> Result['Account', str]:
"""Creates an Account for valid UserSchema's. Or returns a Failure."""
def _create_user(
self, account: 'Account',
) -> Result['User', str]:
"""Create an User instance. If user already exists returns Failure."""
We can implement this feature using a traditional bind
method.
class CreateAccountAndUser(object):
"""Creates new Account-User pair."""
def __call__(self, username: str, email: str) -> Result['User', str]:
"""Can return a Success(user) or Failure(str_reason)."""
return self._validate_user(username, email).bind(
self._create_account,
).bind(
self._create_user,
)
# Protected methods
# ...
And this will work without any problems. But, is it easy to read a code like this? No, it is not.
What alternative we can provide? @pipeline
!
And here’s how we can refactor previous version to be more clear.
class CreateAccountAndUser(object):
"""Creates new Account-User pair."""
@pipeline
def __call__(self, username: str, email: str) -> Result['User', str]:
"""Can return a Success(user) or Failure(str_reason)."""
user_schema = self._validate_user(username, email).unwrap()
account = self._create_account(user_schema).unwrap()
return self._create_user(account)
# Protected methods
# ...
Let’s see how this new .unwrap()
method works:
if you result is Success
it will return its inner value
if your result is Failure
it will raise a UnwrapFailedError
And that’s where @pipeline
decorator becomes in handy.
It will catch any UnwrapFailedError
during the pipeline
and then return a simple Failure
result.
See, do notation allows you to write simple yet powerful pipelines with multiple and complex steps. And at the same time the produced code is simple and readable.
And that’s it!
safe
is used to convert
regular functions that can throw exceptions to functions
that return Result
type.
Supports both async and regular functions.
from returns.result import safe
@safe # Will convert type to: Callable[[int], Result[float, Exception]]
def divide(number: int) -> float:
return number / number
divide(1)
# => Success(1.0)
divide(0)
# => Failure(ZeroDivisionError)
Typing will only work correctly if decorator_plugin is used. This happens due to mypy issue.
Result
(inner_value)[source]¶Bases: returns.primitives.container.GenericContainerTwoSlots
, returns.primitives.container.FixableContainer
, returns.primitives.container.ValueUnwrapContainer
Base class for _Failure and _Success.
map
(function)[source]¶Abstract method to compose container with pure function.
Result
[~_NewValueType, ~_ErrorType]
bind
(function)[source]¶Abstract method to compose container with other container.
Result
[~_NewValueType, ~_NewErrorType]
fix
(function)[source]¶Abstract method to compose container with pure function.
Result
[~_NewValueType, ~_ErrorType]
rescue
(function)[source]¶Abstract method to compose container with other container.
Result
[~_NewValueType, ~_NewErrorType]
Success
(inner_value)[source]¶Public unit function of protected _Success type.
Result
[~_ValueType]