Skip to content

EitherT

funstruct.experimental.monadtransformer.either_t

EitherT — monad transformer that adds typed errors to any monad F.

EitherT[F, E, A] wraps F[Either[E, A]].

Examples:

>>> from funstruct.experimental.monadtransformer.either_t import EitherT
>>> from funstruct.monad.option import Option, Some, Nothing
>>> from funstruct.monad.either import Right, Left

EitherT over Option — combines "might not exist" with "might fail":

>>> EitherT(Some(Right(1))).map(lambda x: x + 10).run()
Some(Right(11))
>>> EitherT(Some(Left("err"))).map(lambda x: x + 10).run()
Some(Left('err'))
>>> EitherT(Nothing()).map(lambda x: x + 10).run()
Nothing()

bind — chains that short-circuit on Left OR Nothing:

>>> inc = lambda x: EitherT(Some(Right(x + 1)))
>>> EitherT(Some(Right(1))).bind(inc).bind(inc).run()
Some(Right(3))
>>> EitherT(Some(Left("stop"))).bind(inc).run()
Some(Left('stop'))

handle_error_with — recover from Left:

>>> EitherT(Some(Left("err"))).handle_error_with(
...     lambda e: EitherT(Some(Right(f"recovered: {e}")))
... ).run()
Some(Right('recovered: err'))

lift_f — bring F[A] into EitherT (wraps value in Right):

>>> EitherT.lift_f(Some(42)).run()
Some(Right(42))

pure — lift_f a plain value into EitherT:

>>> EitherT.pure(99, Option).run()
Some(Right(99))

EitherT

Bases: MonadTransformer, Generic[_F, _E, _A]

EitherT: F[Either[E, A]].

Adds typed error handling to any monad F. Delegates composition to F's bind/map — one implementation for all monads.

Haskell: EitherT e m a (aka ExceptT) Scala: EitherT[F[_], E, A]

Source code in funstruct/experimental/monadtransformer/either_t.py 
 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
128
129
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
170
171
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
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
class EitherT(MonadTransformer, Generic[_F, _E, _A]):
    """EitherT: ``F[Either[E, A]]``.

    Adds typed error handling to any monad F. Delegates composition
    to F's bind/map — one implementation for all monads.

    Haskell: ``EitherT e m a`` (aka ``ExceptT``)
    Scala:   ``EitherT[F[_], E, A]``
    """

    def __init__(self, value) -> None:
        self._value = value

    def run(self):
        """Unwrap to get F[Either[E, A]]."""
        return self._value

    def map(self, f: Callable[[_A], _B]) -> EitherT[_F, _E, _B]:
        """Transform the success value inside F[Either[E, A]]."""
        return EitherT(self._value.map(lambda either: either.map(f)))

    def bind(self, f: Callable[[_A], EitherT[_F, _E, _B]]) -> EitherT[_F, _E, _B]:
        """Chain: unwrap Either inside F, apply f if Right."""

        def _step(either):
            match either:
                case Right(value):
                    return f(value).run()
                case _:
                    return self._value.__class__.pure(either)

        return EitherT(self._value.bind(_step))

    def handle_error_with(
        self, f: Callable[[_E], EitherT[_F, _E, _A]]
    ) -> EitherT[_F, _E, _A]:
        """Recover from Left: f receives the error, returns a new EitherT."""

        def _step(either):
            match either:
                case Left(error):
                    return f(error).run()
                case _:
                    return self._value.__class__.pure(either)

        return EitherT(self._value.bind(_step))

    def left_map(self, f: Callable[[_E], _E]) -> EitherT:
        """Transform the error value. No-op on Right.

        >>> from funstruct.monad.option import Some
        >>> from funstruct.monad.either import Left
        >>> EitherT(Some(Left("err"))).left_map(str.upper).run()
        Some(Left('ERR'))
        """
        return EitherT(self._value.map(lambda either: either.left_map(f)))

    def bimap(self, on_left: Callable, on_right: Callable) -> EitherT:
        """Transform both sides.

        >>> from funstruct.monad.option import Some
        >>> from funstruct.monad.either import Right, Left
        >>> EitherT(Some(Right(5))).bimap(str, lambda x: x * 2).run()
        Some(Right(10))
        >>> EitherT(Some(Left("err"))).bimap(str.upper, lambda x: x * 2).run()
        Some(Left('ERR'))
        """
        return EitherT(self._value.map(lambda either: either.bimap(on_left, on_right)))

    def fold(self, on_left: Callable[[_E], _B], on_right: Callable[[_A], _B]) -> _F:
        """Eliminate the Either inside F, returning F[B].

        >>> from funstruct.monad.option import Some
        >>> from funstruct.monad.either import Right, Left
        >>> EitherT(Some(Right(5))).fold(lambda e: 0, lambda x: x * 2)
        Some(10)
        >>> EitherT(Some(Left("err"))).fold(lambda e: -1, lambda x: x * 2)
        Some(-1)
        """
        return self._value.map(lambda either: either.fold(on_left, on_right))

    def swap(self) -> EitherT:
        """Swap Left and Right inside F.

        >>> from funstruct.monad.option import Some
        >>> from funstruct.monad.either import Right, Left
        >>> EitherT(Some(Right(1))).swap().run()
        Some(Left(1))
        >>> EitherT(Some(Left("err"))).swap().run()
        Some(Right('err'))
        """
        return EitherT(self._value.map(lambda either: either.swap()))

    def get_or_else(self, default: _A) -> _F:
        """Extract the Right value or return default, inside F.

        >>> from funstruct.monad.option import Some
        >>> from funstruct.monad.either import Right, Left
        >>> EitherT(Some(Right(42))).get_or_else(0)
        Some(42)
        >>> EitherT(Some(Left("err"))).get_or_else(0)
        Some(0)
        """
        return self._value.map(lambda either: either.get_or_else(default))

    def and_then(self, other: EitherT) -> EitherT:
        """Kleisli composition: value from self feeds into other's context."""
        return self.bind(lambda _: other)

    @classmethod
    def pure(cls, value: _A, monad: type) -> EitherT:
        """Lift a plain value into EitherT via monad.pure(Right(value))."""
        return cls(monad.pure(Right(value)))

    @classmethod
    def raise_error(cls, error: _E, monad: type) -> EitherT:
        """Lift an error into EitherT via monad.pure(Left(error))."""
        return cls(monad.pure(Left(error)))

    @classmethod
    def lift_f(cls, fa: _F) -> EitherT:
        """Lift F[A] into EitherT — wraps the value in Right.

        Haskell equivalent: ``lift :: m a -> EitherT e m a``
        """
        return cls(fa.map(lambda a: Right(a)))

    @classmethod
    def from_either(cls, either: Either, monad: type) -> EitherT:
        """Lift a plain Either into EitherT."""
        return cls(monad.pure(either))

    @classmethod
    def do(cls, gen_fn) -> Callable[..., EitherT]:
        """Do-notation via generators. Returns a callable.

        Each ``yield`` extracts the Right value from an EitherT.
        Short-circuits on Left (propagated through F).

        >>> from funstruct.monad.option import Some
        >>> from funstruct.monad.either import Right
        >>> def pipeline():
        ...     x = yield EitherT(Some(Right(1)))
        ...     y = yield EitherT(Some(Right(x + 10)))
        ...     return x + y
        >>> EitherT.do(pipeline)().run()
        Some(Right(12))
        """

        def _thunk(*args, **kwargs):
            def _bind_step(either_t, gen):
                def _step(either):
                    match either:
                        case Right(value):
                            try:
                                next_et = gen.send(value)
                                return _bind_step(next_et, gen).run()
                            except StopIteration as e:
                                return either_t.run().__class__.pure(Right(e.value))
                        case _:
                            return either_t.run().__class__.pure(either)

                return EitherT(either_t.run().bind(_step))

            gen = gen_fn(*args, **kwargs)
            try:
                first = next(gen)
            except StopIteration:
                raise ValueError("do block must yield at least once")
            return _bind_step(first, gen)

        return _thunk

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

run()

Unwrap to get F[Either[E, A]].

Source code in funstruct/experimental/monadtransformer/either_t.py 
72
73
74
def run(self):
    """Unwrap to get F[Either[E, A]]."""
    return self._value

map(f)

Transform the success value inside F[Either[E, A]].

Source code in funstruct/experimental/monadtransformer/either_t.py 
76
77
78
def map(self, f: Callable[[_A], _B]) -> EitherT[_F, _E, _B]:
    """Transform the success value inside F[Either[E, A]]."""
    return EitherT(self._value.map(lambda either: either.map(f)))

bind(f)

Chain: unwrap Either inside F, apply f if Right.

Source code in funstruct/experimental/monadtransformer/either_t.py 
80
81
82
83
84
85
86
87
88
89
90
def bind(self, f: Callable[[_A], EitherT[_F, _E, _B]]) -> EitherT[_F, _E, _B]:
    """Chain: unwrap Either inside F, apply f if Right."""

    def _step(either):
        match either:
            case Right(value):
                return f(value).run()
            case _:
                return self._value.__class__.pure(either)

    return EitherT(self._value.bind(_step))

handle_error_with(f)

Recover from Left: f receives the error, returns a new EitherT.

Source code in funstruct/experimental/monadtransformer/either_t.py 
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
def handle_error_with(
    self, f: Callable[[_E], EitherT[_F, _E, _A]]
) -> EitherT[_F, _E, _A]:
    """Recover from Left: f receives the error, returns a new EitherT."""

    def _step(either):
        match either:
            case Left(error):
                return f(error).run()
            case _:
                return self._value.__class__.pure(either)

    return EitherT(self._value.bind(_step))

left_map(f)

Transform the error value. No-op on Right.

from funstruct.monad.option import Some from funstruct.monad.either import Left EitherT(Some(Left("err"))).left_map(str.upper).run() Some(Left('ERR'))

Source code in funstruct/experimental/monadtransformer/either_t.py 
106
107
108
109
110
111
112
113
114
def left_map(self, f: Callable[[_E], _E]) -> EitherT:
    """Transform the error value. No-op on Right.

    >>> from funstruct.monad.option import Some
    >>> from funstruct.monad.either import Left
    >>> EitherT(Some(Left("err"))).left_map(str.upper).run()
    Some(Left('ERR'))
    """
    return EitherT(self._value.map(lambda either: either.left_map(f)))

bimap(on_left, on_right)

Transform both sides.

from funstruct.monad.option import Some from funstruct.monad.either import Right, Left EitherT(Some(Right(5))).bimap(str, lambda x: x * 2).run() Some(Right(10)) EitherT(Some(Left("err"))).bimap(str.upper, lambda x: x * 2).run() Some(Left('ERR'))

Source code in funstruct/experimental/monadtransformer/either_t.py 
116
117
118
119
120
121
122
123
124
125
126
def bimap(self, on_left: Callable, on_right: Callable) -> EitherT:
    """Transform both sides.

    >>> from funstruct.monad.option import Some
    >>> from funstruct.monad.either import Right, Left
    >>> EitherT(Some(Right(5))).bimap(str, lambda x: x * 2).run()
    Some(Right(10))
    >>> EitherT(Some(Left("err"))).bimap(str.upper, lambda x: x * 2).run()
    Some(Left('ERR'))
    """
    return EitherT(self._value.map(lambda either: either.bimap(on_left, on_right)))

fold(on_left, on_right)

Eliminate the Either inside F, returning F[B].

from funstruct.monad.option import Some from funstruct.monad.either import Right, Left EitherT(Some(Right(5))).fold(lambda e: 0, lambda x: x * 2) Some(10) EitherT(Some(Left("err"))).fold(lambda e: -1, lambda x: x * 2) Some(-1)

Source code in funstruct/experimental/monadtransformer/either_t.py 
128
129
130
131
132
133
134
135
136
137
138
def fold(self, on_left: Callable[[_E], _B], on_right: Callable[[_A], _B]) -> _F:
    """Eliminate the Either inside F, returning F[B].

    >>> from funstruct.monad.option import Some
    >>> from funstruct.monad.either import Right, Left
    >>> EitherT(Some(Right(5))).fold(lambda e: 0, lambda x: x * 2)
    Some(10)
    >>> EitherT(Some(Left("err"))).fold(lambda e: -1, lambda x: x * 2)
    Some(-1)
    """
    return self._value.map(lambda either: either.fold(on_left, on_right))

swap()

Swap Left and Right inside F.

from funstruct.monad.option import Some from funstruct.monad.either import Right, Left EitherT(Some(Right(1))).swap().run() Some(Left(1)) EitherT(Some(Left("err"))).swap().run() Some(Right('err'))

Source code in funstruct/experimental/monadtransformer/either_t.py 
140
141
142
143
144
145
146
147
148
149
150
def swap(self) -> EitherT:
    """Swap Left and Right inside F.

    >>> from funstruct.monad.option import Some
    >>> from funstruct.monad.either import Right, Left
    >>> EitherT(Some(Right(1))).swap().run()
    Some(Left(1))
    >>> EitherT(Some(Left("err"))).swap().run()
    Some(Right('err'))
    """
    return EitherT(self._value.map(lambda either: either.swap()))

get_or_else(default)

Extract the Right value or return default, inside F.

from funstruct.monad.option import Some from funstruct.monad.either import Right, Left EitherT(Some(Right(42))).get_or_else(0) Some(42) EitherT(Some(Left("err"))).get_or_else(0) Some(0)

Source code in funstruct/experimental/monadtransformer/either_t.py 
152
153
154
155
156
157
158
159
160
161
162
def get_or_else(self, default: _A) -> _F:
    """Extract the Right value or return default, inside F.

    >>> from funstruct.monad.option import Some
    >>> from funstruct.monad.either import Right, Left
    >>> EitherT(Some(Right(42))).get_or_else(0)
    Some(42)
    >>> EitherT(Some(Left("err"))).get_or_else(0)
    Some(0)
    """
    return self._value.map(lambda either: either.get_or_else(default))

and_then(other)

Kleisli composition: value from self feeds into other's context.

Source code in funstruct/experimental/monadtransformer/either_t.py 
164
165
166
def and_then(self, other: EitherT) -> EitherT:
    """Kleisli composition: value from self feeds into other's context."""
    return self.bind(lambda _: other)

pure(value, monad) classmethod

Lift a plain value into EitherT via monad.pure(Right(value)).

Source code in funstruct/experimental/monadtransformer/either_t.py 
168
169
170
171
@classmethod
def pure(cls, value: _A, monad: type) -> EitherT:
    """Lift a plain value into EitherT via monad.pure(Right(value))."""
    return cls(monad.pure(Right(value)))

raise_error(error, monad) classmethod

Lift an error into EitherT via monad.pure(Left(error)).

Source code in funstruct/experimental/monadtransformer/either_t.py 
173
174
175
176
@classmethod
def raise_error(cls, error: _E, monad: type) -> EitherT:
    """Lift an error into EitherT via monad.pure(Left(error))."""
    return cls(monad.pure(Left(error)))

lift_f(fa) classmethod

Lift F[A] into EitherT — wraps the value in Right.

Haskell equivalent: lift :: m a -> EitherT e m a

Source code in funstruct/experimental/monadtransformer/either_t.py 
178
179
180
181
182
183
184
@classmethod
def lift_f(cls, fa: _F) -> EitherT:
    """Lift F[A] into EitherT — wraps the value in Right.

    Haskell equivalent: ``lift :: m a -> EitherT e m a``
    """
    return cls(fa.map(lambda a: Right(a)))

from_either(either, monad) classmethod

Lift a plain Either into EitherT.

Source code in funstruct/experimental/monadtransformer/either_t.py 
186
187
188
189
@classmethod
def from_either(cls, either: Either, monad: type) -> EitherT:
    """Lift a plain Either into EitherT."""
    return cls(monad.pure(either))

do(gen_fn) classmethod

Do-notation via generators. Returns a callable.

Each yield extracts the Right value from an EitherT. Short-circuits on Left (propagated through F).

from funstruct.monad.option import Some from funstruct.monad.either import Right def pipeline(): ... x = yield EitherT(Some(Right(1))) ... y = yield EitherT(Some(Right(x + 10))) ... return x + y EitherT.do(pipeline)().run() Some(Right(12))

Source code in funstruct/experimental/monadtransformer/either_t.py 
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
218
219
220
221
222
223
224
225
226
227
228
229
230
@classmethod
def do(cls, gen_fn) -> Callable[..., EitherT]:
    """Do-notation via generators. Returns a callable.

    Each ``yield`` extracts the Right value from an EitherT.
    Short-circuits on Left (propagated through F).

    >>> from funstruct.monad.option import Some
    >>> from funstruct.monad.either import Right
    >>> def pipeline():
    ...     x = yield EitherT(Some(Right(1)))
    ...     y = yield EitherT(Some(Right(x + 10)))
    ...     return x + y
    >>> EitherT.do(pipeline)().run()
    Some(Right(12))
    """

    def _thunk(*args, **kwargs):
        def _bind_step(either_t, gen):
            def _step(either):
                match either:
                    case Right(value):
                        try:
                            next_et = gen.send(value)
                            return _bind_step(next_et, gen).run()
                        except StopIteration as e:
                            return either_t.run().__class__.pure(Right(e.value))
                    case _:
                        return either_t.run().__class__.pure(either)

            return EitherT(either_t.run().bind(_step))

        gen = gen_fn(*args, **kwargs)
        try:
            first = next(gen)
        except StopIteration:
            raise ValueError("do block must yield at least once")
        return _bind_step(first, gen)

    return _thunk