Source code for returns._generated.pipeline.pipeline
# -*- coding: utf-8 -*-
from functools import wraps
from inspect import iscoroutinefunction
from returns.primitives.exceptions import UnwrapFailedError
def _pipeline(container_type): # noqa: C901, WPS212
"""
Decorator to enable ``do-notation`` context.
Should be used for series of computations that rely on ``.unwrap`` method.
Supports both async and regular functions.
Works with both ``Maybe`` and ``Result`` containers.
Example:
.. code:: python
>>> from typing import Optional
>>> from returns.pipeline import pipeline
>>> from returns.maybe import Maybe
>>> @pipeline(Maybe)
... def test(one: Optional[int], two: Optional[int]) -> Maybe[int]:
... first = Maybe.new(one).unwrap()
... second = Maybe.new(two).unwrap()
... return Maybe.new(first + second)
...
>>> str(test(1, 2))
'<Some: 3>'
>>> str(test(2, None))
'<Nothing>'
Make sure to supply the correct container type when creating a pipeline.
"""
def factory(function):
if iscoroutinefunction(function):
async def decorator(*args, **kwargs): # noqa: WPS430
try:
return await function(*args, **kwargs)
except UnwrapFailedError as exc:
return exc.halted_container
else:
def decorator(*args, **kwargs): # noqa: WPS430
try:
return function(*args, **kwargs)
except UnwrapFailedError as exc:
return exc.halted_container
return wraps(function)(decorator)
return factory