Skip to content

StateT

funstruct.experimental.monadtransformer.state_t

StateT — state monad transformer over any monad.

Examples:

>>> from funstruct.experimental.monadtransformer import StateT
>>> from funstruct.monad.either import Either, Right, Left
>>> inc = StateT(lambda s: Right((s + 1, s)))
>>> inc.run(0)
Right((1, 0))
>>> StateT.pure(42, Either).run(0)
Right((0, 42))
>>> inc.then(inc).run(0)
Right((2, 1))

StateT

Bases: MonadTransformer, Generic[_F, _A]

Generic state transformer: S -> F[(S, A)].

F is the wrapping monad (Result, FutureResult, Maybe, etc.). Composition uses duck-typed .bind() / .map() / .handle_error_with() on whatever F returns — one implementation for all monads.

Haskell: StateT m s a Scala: StateT[F[_], S, A]

Source code in funstruct/experimental/monadtransformer/state_t.py 
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 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
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
class StateT(MonadTransformer, Generic[_F, _A]):
    """Generic state transformer: ``S -> F[(S, A)]``.

    ``F`` is the wrapping monad (Result, FutureResult, Maybe, etc.).
    Composition uses duck-typed ``.bind()`` / ``.map()`` / ``.handle_error_with()``
    on whatever ``F`` returns — one implementation for all monads.

    Haskell: ``StateT m s a``
    Scala:   ``StateT[F[_], S, A]``
    """

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

    def run(self, initial_state):
        """Execute with initial state.

        Returns ``F[(final_state, value)]``.
        """
        return self._run(initial_state)

    def bind(self, f: Callable[[_A], "StateT[_F, _B]"]) -> "StateT[_F, _B]":
        """FlatMap: thread state, pass value to ``f``.

        >>> from funstruct.monad.option import Option, Some
        >>> StateT.pure(1, Option).bind(lambda x: StateT.pure(x + 10, Option)).run(0)
        Some((0, 11))
        """

        def inner(s):
            return self._run(s).bind(lambda sa: f(sa[1]).run(sa[0]))

        return StateT(inner)

    def map(self, f: Callable[[_A], _B]) -> "StateT[_F, _B]":
        """Transform the produced value without touching state.

        >>> from funstruct.monad.option import Option, Some
        >>> StateT.pure(5, Option).map(lambda x: x * 2).run(0)
        Some((0, 10))
        """

        def inner(s):
            return self._run(s).map(lambda sa: (sa[0], f(sa[1])))

        return StateT(inner)

    def handle_error_with(self, f: Callable[..., "StateT"]) -> "StateT":
        """Recover from failure.

        ``f`` receives the error, returns a recovery StateT.
        Only works when ``F`` supports ``.handle_error_with()``.
        """

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

        return StateT(inner)

    def and_then(self, other: "StateT") -> "StateT":
        """Kleisli composition: value from self becomes initial state for other.

        Short-circuits on inner monad failure.
        """
        return StateT(lambda s: self._run(s).bind(lambda sa: other._run(sa[1])))

    # Constructors — monad class passed explicitly, StateT knows nothing about it

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

        Each `yield` extracts the value from a StateT.
        State threads through, short-circuits on inner monad failure.

        >>> from funstruct.monad.either import Either, Right
        >>> def pipeline():
        ...     x = yield StateT(lambda s: Right((s + 1, s)))
        ...     y = yield StateT(lambda s: Right((s + 1, s)))
        ...     return x + y
        >>> StateT.do(pipeline)().run(0)
        Right((2, 1))
        """

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

                def step(sa):
                    new_s, value = sa
                    try:
                        next_val = gen.send(value)
                        return next_val.run(new_s).bind(step)
                    except StopIteration as e:
                        monad_cls = first.run(s).__class__
                        return _pure(monad_cls, (new_s, e.value))

                return first.run(s).bind(step)

            return cls(_run)

        return _thunk

    @classmethod
    def pure(cls, value, monad: type) -> "StateT":
        """Lift a value into StateT. State unchanged.

        >>> from funstruct.monad.option import Option, Some
        >>> StateT.pure("hello", Option).run(99)
        Some((99, 'hello'))
        """
        return cls(lambda s: _pure(monad, (s, value)))

    @classmethod
    def fail(cls, err: _A, monad: type) -> "StateT":
        """Lift an error. Uses ``monad.raise_error``."""
        return cls(lambda _: monad.raise_error(err))

    @classmethod
    def get(cls, monad: type) -> "StateT":
        """Produce current state as the value.

        >>> from funstruct.monad.option import Option, Some
        >>> StateT.get(Option).run(42)
        Some((42, 42))
        """
        return cls(lambda s: _pure(monad, (s, s)))

    @classmethod
    def set(cls, state, monad: type) -> "StateT":
        """Replace the state entirely, produce None.

        Cats: ``StateT.set``

        >>> from funstruct.monad.option import Option, Some
        >>> StateT.set(99, Option).run(0)
        Some((99, None))
        """
        return cls(lambda _: _pure(monad, (state, None)))

    @classmethod
    def inspect(cls, f: Callable, monad: type) -> "StateT":
        """Get a function of the state as the value, without modifying state.

        Cats: ``StateT.inspect``

        >>> from funstruct.monad.option import Option, Some
        >>> StateT.inspect(lambda s: s * 2, Option).run(5)
        Some((5, 10))
        """
        return cls(lambda s: _pure(monad, (s, f(s))))

    @classmethod
    def modify(cls, f: Callable[..., object], monad: type) -> "StateT":
        """Modify state, produce None.

        >>> from funstruct.monad.option import Option, Some
        >>> StateT.modify(lambda s: s + 1, Option).run(5)
        Some((6, None))
        """
        return cls(lambda s: _pure(monad, (f(s), None)))

    @classmethod
    def lift_f(cls, inner: _F) -> "StateT":
        """Lift F[A] into StateT — state unchanged.

        Haskell equivalent: ``lift :: m a -> StateT s m a``

        >>> from funstruct.monad.option import Option, Some, Nothing
        >>> StateT.lift_f(Some(42)).run(0)
        Some((0, 42))
        >>> StateT.lift_f(Nothing()).run(0)
        Nothing()
        """
        return cls(lambda s: inner.map(lambda a: (s, a)))

    @classmethod
    def from_state(cls, state: Callable[..., tuple], monad: type) -> "StateT":
        """Lift a pure State (S → (S, A)) into StateT.

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

        >>> from funstruct.monad.option import Option, Some
        >>> StateT.from_state(lambda s: (s + 1, s), Option).run(0)
        Some((1, 0))
        """
        return cls(lambda s: _pure(monad, state(s)))

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

run(initial_state)

Execute with initial state.

Returns F[(final_state, value)].

Source code in funstruct/experimental/monadtransformer/state_t.py 
47
48
49
50
51
52
def run(self, initial_state):
    """Execute with initial state.

    Returns ``F[(final_state, value)]``.
    """
    return self._run(initial_state)

bind(f)

FlatMap: thread state, pass value to f.

from funstruct.monad.option import Option, Some StateT.pure(1, Option).bind(lambda x: StateT.pure(x + 10, Option)).run(0) Some((0, 11))

Source code in funstruct/experimental/monadtransformer/state_t.py 
54
55
56
57
58
59
60
61
62
63
64
65
def bind(self, f: Callable[[_A], "StateT[_F, _B]"]) -> "StateT[_F, _B]":
    """FlatMap: thread state, pass value to ``f``.

    >>> from funstruct.monad.option import Option, Some
    >>> StateT.pure(1, Option).bind(lambda x: StateT.pure(x + 10, Option)).run(0)
    Some((0, 11))
    """

    def inner(s):
        return self._run(s).bind(lambda sa: f(sa[1]).run(sa[0]))

    return StateT(inner)

map(f)

Transform the produced value without touching state.

from funstruct.monad.option import Option, Some StateT.pure(5, Option).map(lambda x: x * 2).run(0) Some((0, 10))

Source code in funstruct/experimental/monadtransformer/state_t.py 
67
68
69
70
71
72
73
74
75
76
77
78
def map(self, f: Callable[[_A], _B]) -> "StateT[_F, _B]":
    """Transform the produced value without touching state.

    >>> from funstruct.monad.option import Option, Some
    >>> StateT.pure(5, Option).map(lambda x: x * 2).run(0)
    Some((0, 10))
    """

    def inner(s):
        return self._run(s).map(lambda sa: (sa[0], f(sa[1])))

    return StateT(inner)

handle_error_with(f)

Recover from failure.

f receives the error, returns a recovery StateT. Only works when F supports .handle_error_with().

Source code in funstruct/experimental/monadtransformer/state_t.py 
80
81
82
83
84
85
86
87
88
89
90
def handle_error_with(self, f: Callable[..., "StateT"]) -> "StateT":
    """Recover from failure.

    ``f`` receives the error, returns a recovery StateT.
    Only works when ``F`` supports ``.handle_error_with()``.
    """

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

    return StateT(inner)

and_then(other)

Kleisli composition: value from self becomes initial state for other.

Short-circuits on inner monad failure.

Source code in funstruct/experimental/monadtransformer/state_t.py 
92
93
94
95
96
97
def and_then(self, other: "StateT") -> "StateT":
    """Kleisli composition: value from self becomes initial state for other.

    Short-circuits on inner monad failure.
    """
    return StateT(lambda s: self._run(s).bind(lambda sa: other._run(sa[1])))

do(gen_fn) classmethod

Do-notation via generators. Returns a callable.

Each yield extracts the value from a StateT. State threads through, short-circuits on inner monad failure.

from funstruct.monad.either import Either, Right def pipeline(): ... x = yield StateT(lambda s: Right((s + 1, s))) ... y = yield StateT(lambda s: Right((s + 1, s))) ... return x + y StateT.do(pipeline)().run(0) Right((2, 1))

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

    Each `yield` extracts the value from a StateT.
    State threads through, short-circuits on inner monad failure.

    >>> from funstruct.monad.either import Either, Right
    >>> def pipeline():
    ...     x = yield StateT(lambda s: Right((s + 1, s)))
    ...     y = yield StateT(lambda s: Right((s + 1, s)))
    ...     return x + y
    >>> StateT.do(pipeline)().run(0)
    Right((2, 1))
    """

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

            def step(sa):
                new_s, value = sa
                try:
                    next_val = gen.send(value)
                    return next_val.run(new_s).bind(step)
                except StopIteration as e:
                    monad_cls = first.run(s).__class__
                    return _pure(monad_cls, (new_s, e.value))

            return first.run(s).bind(step)

        return cls(_run)

    return _thunk

pure(value, monad) classmethod

Lift a value into StateT. State unchanged.

from funstruct.monad.option import Option, Some StateT.pure("hello", Option).run(99) Some((99, 'hello'))

Source code in funstruct/experimental/monadtransformer/state_t.py 
140
141
142
143
144
145
146
147
148
@classmethod
def pure(cls, value, monad: type) -> "StateT":
    """Lift a value into StateT. State unchanged.

    >>> from funstruct.monad.option import Option, Some
    >>> StateT.pure("hello", Option).run(99)
    Some((99, 'hello'))
    """
    return cls(lambda s: _pure(monad, (s, value)))

fail(err, monad) classmethod

Lift an error. Uses monad.raise_error.

Source code in funstruct/experimental/monadtransformer/state_t.py 
150
151
152
153
@classmethod
def fail(cls, err: _A, monad: type) -> "StateT":
    """Lift an error. Uses ``monad.raise_error``."""
    return cls(lambda _: monad.raise_error(err))

get(monad) classmethod

Produce current state as the value.

from funstruct.monad.option import Option, Some StateT.get(Option).run(42) Some((42, 42))

Source code in funstruct/experimental/monadtransformer/state_t.py 
155
156
157
158
159
160
161
162
163
@classmethod
def get(cls, monad: type) -> "StateT":
    """Produce current state as the value.

    >>> from funstruct.monad.option import Option, Some
    >>> StateT.get(Option).run(42)
    Some((42, 42))
    """
    return cls(lambda s: _pure(monad, (s, s)))

set(state, monad) classmethod

Replace the state entirely, produce None.

Cats: StateT.set

from funstruct.monad.option import Option, Some StateT.set(99, Option).run(0) Some((99, None))

Source code in funstruct/experimental/monadtransformer/state_t.py 
165
166
167
168
169
170
171
172
173
174
175
@classmethod
def set(cls, state, monad: type) -> "StateT":
    """Replace the state entirely, produce None.

    Cats: ``StateT.set``

    >>> from funstruct.monad.option import Option, Some
    >>> StateT.set(99, Option).run(0)
    Some((99, None))
    """
    return cls(lambda _: _pure(monad, (state, None)))

inspect(f, monad) classmethod

Get a function of the state as the value, without modifying state.

Cats: StateT.inspect

from funstruct.monad.option import Option, Some StateT.inspect(lambda s: s * 2, Option).run(5) Some((5, 10))

Source code in funstruct/experimental/monadtransformer/state_t.py 
177
178
179
180
181
182
183
184
185
186
187
@classmethod
def inspect(cls, f: Callable, monad: type) -> "StateT":
    """Get a function of the state as the value, without modifying state.

    Cats: ``StateT.inspect``

    >>> from funstruct.monad.option import Option, Some
    >>> StateT.inspect(lambda s: s * 2, Option).run(5)
    Some((5, 10))
    """
    return cls(lambda s: _pure(monad, (s, f(s))))

modify(f, monad) classmethod

Modify state, produce None.

from funstruct.monad.option import Option, Some StateT.modify(lambda s: s + 1, Option).run(5) Some((6, None))

Source code in funstruct/experimental/monadtransformer/state_t.py 
189
190
191
192
193
194
195
196
197
@classmethod
def modify(cls, f: Callable[..., object], monad: type) -> "StateT":
    """Modify state, produce None.

    >>> from funstruct.monad.option import Option, Some
    >>> StateT.modify(lambda s: s + 1, Option).run(5)
    Some((6, None))
    """
    return cls(lambda s: _pure(monad, (f(s), None)))

lift_f(inner) classmethod

Lift F[A] into StateT — state unchanged.

Haskell equivalent: lift :: m a -> StateT s m a

from funstruct.monad.option import Option, Some, Nothing StateT.lift_f(Some(42)).run(0) Some((0, 42)) StateT.lift_f(Nothing()).run(0) Nothing()

Source code in funstruct/experimental/monadtransformer/state_t.py 
199
200
201
202
203
204
205
206
207
208
209
210
211
@classmethod
def lift_f(cls, inner: _F) -> "StateT":
    """Lift F[A] into StateT — state unchanged.

    Haskell equivalent: ``lift :: m a -> StateT s m a``

    >>> from funstruct.monad.option import Option, Some, Nothing
    >>> StateT.lift_f(Some(42)).run(0)
    Some((0, 42))
    >>> StateT.lift_f(Nothing()).run(0)
    Nothing()
    """
    return cls(lambda s: inner.map(lambda a: (s, a)))

from_state(state, monad) classmethod

Lift a pure State (S → (S, A)) into StateT.

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

from funstruct.monad.option import Option, Some StateT.from_state(lambda s: (s + 1, s), Option).run(0) Some((1, 0))

Source code in funstruct/experimental/monadtransformer/state_t.py 
213
214
215
216
217
218
219
220
221
222
223
@classmethod
def from_state(cls, state: Callable[..., tuple], monad: type) -> "StateT":
    """Lift a pure State (S → (S, A)) into StateT.

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

    >>> from funstruct.monad.option import Option, Some
    >>> StateT.from_state(lambda s: (s + 1, s), Option).run(0)
    Some((1, 0))
    """
    return cls(lambda s: _pure(monad, state(s)))