Skip to content

Either

funstruct.monad.either

Either monad — typed error handling.

Either[E, A] = Right(a) | Left(e). Right-biased.

Examples:

>>> from funstruct.monad.either import Either, Right, Left
>>> Right(10).map(lambda x: x + 1)
Right(11)
>>> Left("err").map(lambda x: x + 1)
Left('err')
>>> Right(10).bind(lambda x: Right(x * 2))
Right(20)
>>> Left("err").bind(lambda x: Right(x * 2))
Left('err')

handle_error_with — recover from Left:

>>> Left("err").handle_error_with(lambda e: Right("default"))
Right('default')
>>> Right(10).handle_error_with(lambda e: Right("default"))
Right(10)

do-notation:

>>> def pipeline():
...     x = yield Right(1)
...     y = yield Right(x + 10)
...     return x + y
>>> Either.do(pipeline)()
Right(12)

Either

Bases: DataType, Generic[E, A]

Either[E, A]: Right(value) or Left(error).

Right-biased monad. bind/map/>> operate on the Right value and short-circuit on Left.

Source code in funstruct/monad/either/__init__.py 
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
class Either(DataType, Generic[E, A]):
    """Either[E, A]: Right(value) or Left(error).

    Right-biased monad. bind/map/>> operate on the Right value
    and short-circuit on Left.
    """

    @classmethod
    def pure(cls, value: A) -> Either[E, A]:
        return Right(value)

    @classmethod
    def raise_error(cls, error: E) -> Either[E, A]:
        return Left(error)

    @classmethod
    def do(cls, gen_fn: Callable) -> Callable[..., Either]:
        """Do-notation. Short-circuits on Left. Returns a callable.

        # TODO: do-notation is ~2x slower than raw bind chains due to generator
        # protocol overhead. Consider optimizing the generator loop or providing
        # a bind-chain builder as an alternative for performance-sensitive code.

        >>> def pipeline():
        ...     x = yield Right(1)
        ...     y = yield Right(x + 10)
        ...     return x + y
        >>> Either.do(pipeline)()
        Right(12)
        """

        def _thunk(*args, **kwargs):
            gen = gen_fn(*args, **kwargs)
            try:
                monadic_val = next(gen)
                while True:
                    match monadic_val:
                        case Left():
                            return monadic_val
                        case Right(value):
                            monadic_val = gen.send(value)
            except StopIteration as e:
                return Right(e.value)

        return _thunk

    @classmethod
    def sequence(cls, eithers: CList[Either[E, A]]) -> Either[E, CList[A]]:
        """CList[Either[E, A]] -> Either[E, CList[A]]."""
        from funstruct.collections.cons import Cons, Nil
        from funstruct.util.tailrec import tail_call, tco

        @tco
        def _go(remaining, acc):
            match remaining:
                case Nil():
                    return Right(acc.reversed())
                case Cons(head, tail):
                    match head:
                        case Left():
                            return head
                        case Right(v):
                            return tail_call(_go)(tail, Cons(v, acc))

        return _go(eithers, Nil())

    @classmethod
    def traverse(cls, values, f: Callable[[A], Either[E, B]]) -> Either:
        return cls.sequence(values.map(f))

    @property
    def is_right(self) -> bool:
        return False

    @property
    def is_left(self) -> bool:
        return not self.is_right

do(gen_fn) classmethod

Do-notation. Short-circuits on Left. Returns a callable.

TODO: do-notation is ~2x slower than raw bind chains due to generator
protocol overhead. Consider optimizing the generator loop or providing
a bind-chain builder as an alternative for performance-sensitive code.

def pipeline(): ... x = yield Right(1) ... y = yield Right(x + 10) ... return x + y Either.do(pipeline)() Right(12)

Source code in funstruct/monad/either/__init__.py 
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
@classmethod
def do(cls, gen_fn: Callable) -> Callable[..., Either]:
    """Do-notation. Short-circuits on Left. Returns a callable.

    # TODO: do-notation is ~2x slower than raw bind chains due to generator
    # protocol overhead. Consider optimizing the generator loop or providing
    # a bind-chain builder as an alternative for performance-sensitive code.

    >>> def pipeline():
    ...     x = yield Right(1)
    ...     y = yield Right(x + 10)
    ...     return x + y
    >>> Either.do(pipeline)()
    Right(12)
    """

    def _thunk(*args, **kwargs):
        gen = gen_fn(*args, **kwargs)
        try:
            monadic_val = next(gen)
            while True:
                match monadic_val:
                    case Left():
                        return monadic_val
                    case Right(value):
                        monadic_val = gen.send(value)
        except StopIteration as e:
            return Right(e.value)

    return _thunk

sequence(eithers) classmethod

CList[Either[E, A]] -> Either[E, CList[A]].

Source code in funstruct/monad/either/__init__.py 
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
@classmethod
def sequence(cls, eithers: CList[Either[E, A]]) -> Either[E, CList[A]]:
    """CList[Either[E, A]] -> Either[E, CList[A]]."""
    from funstruct.collections.cons import Cons, Nil
    from funstruct.util.tailrec import tail_call, tco

    @tco
    def _go(remaining, acc):
        match remaining:
            case Nil():
                return Right(acc.reversed())
            case Cons(head, tail):
                match head:
                    case Left():
                        return head
                    case Right(v):
                        return tail_call(_go)(tail, Cons(v, acc))

    return _go(eithers, Nil())

Right dataclass

Bases: Either[E, A]

Success case.

Source code in funstruct/monad/either/__init__.py 
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
@dataclass(frozen=True, eq=False)
class Right(Either[E, A]):
    """Success case."""

    value: A

    @property
    def is_right(self) -> bool:
        return True

    def bind(self, f: Callable[[A], Either[E, B]]) -> Either[E, B]:
        return f(self.value)

    def left_map(self, f: Callable[[E], E]) -> Either[E, A]:
        return self

    def handle_error_with(self, f: Callable[[E], Either]) -> Either[E, A]:
        return self

    def bimap(self, on_left: Callable[[E], E], on_right: Callable[[A], B]) -> Either:
        return Right(on_right(self.value))

    def get_or_else(self, default: A) -> A:
        return self.value

    def fold(self, on_left: Callable[[E], C], on_right: Callable[[A], C]) -> C:
        return on_right(self.value)

    def swap(self) -> Either[A, E]:
        return Left(self.value)

    def __eq__(self, other: object) -> bool:
        match other:
            case Right(val):
                return self.value == val
            case _:
                return False

    def __repr__(self) -> str:
        return f"Right({repr(self.value)})"

Left dataclass

Bases: CapturesCreationSiteMixin, Either[E, A]

Error case. Captures creation site automatically.

Source code in funstruct/monad/either/__init__.py 
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
@dataclass(frozen=True, eq=False)
class Left(CapturesCreationSiteMixin, Either[E, A]):
    """Error case. Captures creation site automatically."""

    error: E

    @property
    def is_right(self) -> bool:
        return False

    def bind(self, f: Callable[[A], Either[E, B]]) -> Either[E, B]:
        return self

    def left_map(self, f: Callable[[E], E]) -> Either[E, A]:
        """>>> Left("oops").left_map(lambda e: e.upper())
        Left('OOPS')
        """
        return Left(f(self.error))

    def handle_error_with(self, f: Callable[[E], Either]) -> Either:
        """>>> Left("oops").handle_error_with(lambda e: Right(f"recovered: {e}"))
        Right('recovered: oops')
        """
        return f(self.error)

    def bimap(self, on_left: Callable[[E], E], on_right: Callable[[A], B]) -> Either:
        return Left(on_left(self.error))

    def get_or_else(self, default: A) -> A:
        return default

    def fold(self, on_left: Callable[[E], C], on_right: Callable[[A], C]) -> C:
        return on_left(self.error)

    def swap(self) -> Either[A, E]:
        return Right(self.error)

    def __eq__(self, other: object) -> bool:
        match other:
            case Left(err):
                return self.error == err
            case _:
                return False

    def __repr__(self) -> str:
        return f"Left({repr(self.error)})"

left_map(f)

Left("oops").left_map(lambda e: e.upper()) Left('OOPS')

Source code in funstruct/monad/either/__init__.py 
185
186
187
188
189
def left_map(self, f: Callable[[E], E]) -> Either[E, A]:
    """>>> Left("oops").left_map(lambda e: e.upper())
    Left('OOPS')
    """
    return Left(f(self.error))

handle_error_with(f)

Left("oops").handle_error_with(lambda e: Right(f"recovered: {e}")) Right('recovered: oops')

Source code in funstruct/monad/either/__init__.py 
191
192
193
194
195
def handle_error_with(self, f: Callable[[E], Either]) -> Either:
    """>>> Left("oops").handle_error_with(lambda e: Right(f"recovered: {e}"))
    Right('recovered: oops')
    """
    return f(self.error)