Skip to content

ReaderT

funstruct.experimental.monadtransformer.reader_t

ReaderT — reader monad transformer over any monad.

ReaderT — shared environment + inner monad's effects (failure, state, etc.)

ReaderT[F, Ctx, A]  =  Ctx -> F[A]

bind: chain computations that share context, any can fail. and_then: pipe output forward as the next context (Kleisli composition).

Examples:

>>> from funstruct.experimental.monadtransformer import ReaderT
>>> from funstruct.monad.either import Either, Right, Left

bind — shared context, with failure:

>>> get_db = ReaderT(lambda cfg: (
...     Right(cfg["db"]) if "db" in cfg
...     else Left("missing db")))
>>> validate = lambda url: ReaderT(lambda cfg: (
...     Right(url) if url.startswith("postgres://")
...     else Left(f"bad url: {url}")))
>>> connect = lambda url: ReaderT(lambda cfg: (
...     Right(f"{url} as {cfg['user']}")))
>>> pipeline = (
...     get_db
...     .bind(validate)
...     .bind(connect)
... )
>>> pipeline.run({"db": "postgres://localhost/app", "user": "admin"})
Right('postgres://localhost/app as admin')
>>> pipeline.run({"db": "mysql://bad", "user": "admin"})
Left('bad url: mysql://bad')

and_then — output feeds as next input, short-circuits on failure:

>>> parse_int = ReaderT(lambda s: (
...     Right(int(s)) if s.isdigit()
...     else Left(f"not a number: {s}")))
>>> double = ReaderT(lambda n: Right(n * 2))
>>> to_str = ReaderT(lambda n: Right(str(n)))
>>> pipeline = (
...     parse_int
...     .and_then(double)
...     .and_then(to_str)
... )
>>> pipeline.run("21")
Right('42')
>>> pipeline.run("abc")
Left('not a number: abc')

ReaderT

Bases: MonadTransformer, Generic[_Ctx, _M, _A]

ReaderT: Ctx -> M[A].

M is the inner monad (StateT, Result, etc.). Composition delegates to M's bind/map/lash — ReaderT just threads the context to both sides.

Source code in funstruct/experimental/monadtransformer/reader_t.py 
 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
class ReaderT(MonadTransformer, Generic[_Ctx, _M, _A]):
    """ReaderT: ``Ctx -> M[A]``.

    M is the inner monad (StateT, Result, etc.). Composition delegates to M's
    bind/map/lash — ReaderT just threads the context to both sides.
    """

    def __init__(self, run: Callable[[_Ctx], _M]) -> None:
        self._run = run

    def run(self, ctx):
        """Apply context, returning the inner M[A]."""
        return self._run(ctx)

    def __call__(self, ctx):
        """Apply context (alias for run)."""
        return self._run(ctx)

    def bind(
        self,
        f: Callable[[_A], ReaderT[_Ctx, _M, _B]],
    ) -> ReaderT[_Ctx, _M, _B]:
        """FlatMap: compose via M's bind, threading ctx."""

        def inner(ctx):
            return self._run(ctx).bind(lambda a: f(a).run(ctx))

        return ReaderT(inner)

    def map(self, f: Callable[[_A], _B]) -> ReaderT[_Ctx, _M, _B]:
        """Transform value via M's map."""

        def inner(ctx):
            return self._run(ctx).map(f)

        return ReaderT(inner)

    def handle_error_with(self, f: Callable[..., ReaderT]) -> ReaderT[_Ctx, _M, _A]:
        """Recover from failure via inner monad's handle_error_with."""

        def inner(ctx):
            return self._run(ctx).handle_error_with(lambda err: f(err).run(ctx))

        return ReaderT(inner)

    def local(self, f: Callable) -> ReaderT:
        """Transform the environment before running.

        Cats: ``Kleisli.local``

        >>> from funstruct.monad.either import Right
        >>> r = ReaderT(lambda ctx: Right(ctx["name"]))
        >>> r.local(lambda outer: {"name": outer}).run("Alice")
        Right('Alice')
        """
        return ReaderT(lambda ctx: self._run(f(ctx)))

    @classmethod
    def ask(cls, monad: type) -> ReaderT:
        """Get the environment as the value.

        Cats: ``Kleisli.ask``

        >>> from funstruct.monad.either import Either, Right
        >>> ReaderT.ask(Either).run(42)
        Right(42)
        """
        return cls(lambda ctx: _pure(monad, ctx))

    # ap inherited from MonadTransformer (derived from bind + map)

    def and_then(self, other: ReaderT) -> ReaderT:
        """Kleisli composition: output of self becomes input (ctx) of other.

        Short-circuits on inner monad failure.
        """
        return ReaderT(
            lambda ctx: self._run(ctx).bind(lambda result: other._run(result))
        )

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

        Each `yield` extracts the value from a ReaderT (shared ctx).
        Short-circuits on inner monad failure.

        >>> from funstruct.monad.either import Right
        >>> def pipeline():
        ...     x = yield ReaderT(lambda ctx: Right(ctx))
        ...     y = yield ReaderT(lambda ctx: Right(x + ctx))
        ...     return y
        >>> ReaderT.do(pipeline)().run(5)
        Right(10)
        """

        def _thunk(*args, **kwargs):
            def _run(ctx):
                gen = gen_fn(*args, **kwargs)
                try:
                    monadic_val = next(gen)
                except StopIteration:
                    raise ValueError("do block must yield at least once")

                def step(value):
                    try:
                        next_val = gen.send(value)
                        return next_val._run(ctx).bind(step)
                    except StopIteration as e:
                        return _pure(monadic_val._run(ctx).__class__, e.value)

                return monadic_val._run(ctx).bind(step)

            return cls(_run)

        return _thunk

    @classmethod
    def pure(cls, value: _A, monad: type) -> ReaderT:
        """Lift a plain value into ReaderT via monad.pure."""
        return cls(lambda _: _pure(monad, value))

    @classmethod
    def from_reader(cls, reader_fn: Callable[[_Ctx], _A], monad: type) -> ReaderT:
        """Lift a pure Reader (Ctx → A) into ReaderT.

        Use when you have the inner reader function but not the outer monad.

        >>> from funstruct.monad.either import Either, Right
        >>> ReaderT.from_reader(lambda ctx: ctx["name"], Either).run({"name": "Alice"})
        Right('Alice')
        """
        return cls(lambda ctx: _pure(monad, reader_fn(ctx)))

    @classmethod
    def lift_f(cls, m: _M) -> ReaderT:
        """Lift M[A] into ReaderT (ignoring context).

        Haskell equivalent: ``lift :: m a -> ReaderT r m a``
        """
        return cls(lambda _: m)

    def __repr__(self) -> str:
        return f"ReaderT({self._run})"

run(ctx)

Apply context, returning the inner M[A].

Source code in funstruct/experimental/monadtransformer/reader_t.py 
83
84
85
def run(self, ctx):
    """Apply context, returning the inner M[A]."""
    return self._run(ctx)

__call__(ctx)

Apply context (alias for run).

Source code in funstruct/experimental/monadtransformer/reader_t.py 
87
88
89
def __call__(self, ctx):
    """Apply context (alias for run)."""
    return self._run(ctx)

bind(f)

FlatMap: compose via M's bind, threading ctx.

Source code in funstruct/experimental/monadtransformer/reader_t.py 
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
def bind(
    self,
    f: Callable[[_A], ReaderT[_Ctx, _M, _B]],
) -> ReaderT[_Ctx, _M, _B]:
    """FlatMap: compose via M's bind, threading ctx."""

    def inner(ctx):
        return self._run(ctx).bind(lambda a: f(a).run(ctx))

    return ReaderT(inner)

map(f)

Transform value via M's map.

Source code in funstruct/experimental/monadtransformer/reader_t.py 
102
103
104
105
106
107
108
def map(self, f: Callable[[_A], _B]) -> ReaderT[_Ctx, _M, _B]:
    """Transform value via M's map."""

    def inner(ctx):
        return self._run(ctx).map(f)

    return ReaderT(inner)

handle_error_with(f)

Recover from failure via inner monad's handle_error_with.

Source code in funstruct/experimental/monadtransformer/reader_t.py 
110
111
112
113
114
115
116
def handle_error_with(self, f: Callable[..., ReaderT]) -> ReaderT[_Ctx, _M, _A]:
    """Recover from failure via inner monad's handle_error_with."""

    def inner(ctx):
        return self._run(ctx).handle_error_with(lambda err: f(err).run(ctx))

    return ReaderT(inner)

local(f)

Transform the environment before running.

Cats: Kleisli.local

from funstruct.monad.either import Right r = ReaderT(lambda ctx: Right(ctx["name"])) r.local(lambda outer: {"name": outer}).run("Alice") Right('Alice')

Source code in funstruct/experimental/monadtransformer/reader_t.py 
118
119
120
121
122
123
124
125
126
127
128
def local(self, f: Callable) -> ReaderT:
    """Transform the environment before running.

    Cats: ``Kleisli.local``

    >>> from funstruct.monad.either import Right
    >>> r = ReaderT(lambda ctx: Right(ctx["name"]))
    >>> r.local(lambda outer: {"name": outer}).run("Alice")
    Right('Alice')
    """
    return ReaderT(lambda ctx: self._run(f(ctx)))

ask(monad) classmethod

Get the environment as the value.

Cats: Kleisli.ask

from funstruct.monad.either import Either, Right ReaderT.ask(Either).run(42) Right(42)

Source code in funstruct/experimental/monadtransformer/reader_t.py 
130
131
132
133
134
135
136
137
138
139
140
@classmethod
def ask(cls, monad: type) -> ReaderT:
    """Get the environment as the value.

    Cats: ``Kleisli.ask``

    >>> from funstruct.monad.either import Either, Right
    >>> ReaderT.ask(Either).run(42)
    Right(42)
    """
    return cls(lambda ctx: _pure(monad, ctx))

and_then(other)

Kleisli composition: output of self becomes input (ctx) of other.

Short-circuits on inner monad failure.

Source code in funstruct/experimental/monadtransformer/reader_t.py 
144
145
146
147
148
149
150
151
def and_then(self, other: ReaderT) -> ReaderT:
    """Kleisli composition: output of self becomes input (ctx) of other.

    Short-circuits on inner monad failure.
    """
    return ReaderT(
        lambda ctx: self._run(ctx).bind(lambda result: other._run(result))
    )

do(gen_fn) classmethod

Do-notation via generators. Returns a callable.

Each yield extracts the value from a ReaderT (shared ctx). Short-circuits on inner monad failure.

from funstruct.monad.either import Right def pipeline(): ... x = yield ReaderT(lambda ctx: Right(ctx)) ... y = yield ReaderT(lambda ctx: Right(x + ctx)) ... return y ReaderT.do(pipeline)().run(5) Right(10)

Source code in funstruct/experimental/monadtransformer/reader_t.py 
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
@classmethod
def do(cls, gen_fn) -> Callable[..., ReaderT]:
    """Do-notation via generators. Returns a callable.

    Each `yield` extracts the value from a ReaderT (shared ctx).
    Short-circuits on inner monad failure.

    >>> from funstruct.monad.either import Right
    >>> def pipeline():
    ...     x = yield ReaderT(lambda ctx: Right(ctx))
    ...     y = yield ReaderT(lambda ctx: Right(x + ctx))
    ...     return y
    >>> ReaderT.do(pipeline)().run(5)
    Right(10)
    """

    def _thunk(*args, **kwargs):
        def _run(ctx):
            gen = gen_fn(*args, **kwargs)
            try:
                monadic_val = next(gen)
            except StopIteration:
                raise ValueError("do block must yield at least once")

            def step(value):
                try:
                    next_val = gen.send(value)
                    return next_val._run(ctx).bind(step)
                except StopIteration as e:
                    return _pure(monadic_val._run(ctx).__class__, e.value)

            return monadic_val._run(ctx).bind(step)

        return cls(_run)

    return _thunk

pure(value, monad) classmethod

Lift a plain value into ReaderT via monad.pure.

Source code in funstruct/experimental/monadtransformer/reader_t.py 
190
191
192
193
@classmethod
def pure(cls, value: _A, monad: type) -> ReaderT:
    """Lift a plain value into ReaderT via monad.pure."""
    return cls(lambda _: _pure(monad, value))

from_reader(reader_fn, monad) classmethod

Lift a pure Reader (Ctx → A) into ReaderT.

Use when you have the inner reader function but not the outer monad.

from funstruct.monad.either import Either, Right ReaderT.from_reader(lambda ctx: ctx["name"], Either).run({"name": "Alice"}) Right('Alice')

Source code in funstruct/experimental/monadtransformer/reader_t.py 
195
196
197
198
199
200
201
202
203
204
205
@classmethod
def from_reader(cls, reader_fn: Callable[[_Ctx], _A], monad: type) -> ReaderT:
    """Lift a pure Reader (Ctx → A) into ReaderT.

    Use when you have the inner reader function but not the outer monad.

    >>> from funstruct.monad.either import Either, Right
    >>> ReaderT.from_reader(lambda ctx: ctx["name"], Either).run({"name": "Alice"})
    Right('Alice')
    """
    return cls(lambda ctx: _pure(monad, reader_fn(ctx)))

lift_f(m) classmethod

Lift M[A] into ReaderT (ignoring context).

Haskell equivalent: lift :: m a -> ReaderT r m a

Source code in funstruct/experimental/monadtransformer/reader_t.py 
207
208
209
210
211
212
213
@classmethod
def lift_f(cls, m: _M) -> ReaderT:
    """Lift M[A] into ReaderT (ignoring context).

    Haskell equivalent: ``lift :: m a -> ReaderT r m a``
    """
    return cls(lambda _: m)