monawhat package

Submodules

monawhat.base module

Module implementing a base monad class that all other monads can extend.

class monawhat.base.BaseMonad

Bases: Generic[A]

Base class for all monads.

This class provides common functionality for monads and serves as a foundation for implementing specific monads. It uses the Identity monad internally for pure values.

Subclasses must implement _bind_implementation and _map_implementation.

bind(f: Callable[[A], Any]) → Any

Chain this monad with a function that returns another monad.

Parameters:

f – A function that takes the value(s) of this monad and returns a new monad.

Returns:

The new monad returned by the function.

map(f: Callable[[A], B]) → BaseMonad[B]

Apply a function to the value(s) inside the monad.

Parameters:

f – The function to apply to the value(s).

Returns:

A new monad with the transformed value(s).

classmethod pure(value: A, *args: Any, **kwargs: Any) → BaseMonad[A]

Create a monad containing a pure value.

This uses the Identity monad internally to represent pure values. Subclasses should override this to provide monad-specific implementation.

Parameters:

• value – The value to wrap in the monad.

• *args – Additional positional arguments for specific monad implementations.

• **kwargs – Additional keyword arguments for specific monad implementations.

Returns:

A monad containing the value.

monawhat.either module

Module implementing the Either monad for handling success/failure cases.

Examples

Basic usage for error handling:

>>> from monawhat.either import Either
>>>
>>> def safe_divide(a, b):
...     if b == 0:
...         return Either.left(f"Cannot divide {a} by zero")
...     return Either.right(a / b)
>>>
>>> # Successful case
>>> result = safe_divide(10, 2)
>>> if result.is_right():
...     print(f"Result: {result.get()}")  # Output: Result: 5.0
>>>
>>> # Error case
>>> result = safe_divide(10, 0)
>>> if result.is_left():
...     print(f"Error: {result.get()}")  # Output: Error: Cannot divide 10 by zero

Chaining operations with bind:

>>> def get_user(user_id):
...     # Simulate fetching a user from database
...     if user_id > 0:
...         return Either.right({"id": user_id, "email": "user@example.com"})
...     return Either.left(f"User with id {user_id} not found")
>>>
>>> def send_email(user, message):
...     # Simulate sending an email
...     if "email" in user:
...         return Either.right(f"Email sent to {user['email']}")
...     return Either.left("Email address not found")
>>>
>>> # Chain operations together
>>> result = (Either.right(1)
...           .bind(get_user)
...           .bind(lambda user: send_email(user, "Welcome!")))
>>> print(result)  # Output: Right(Email sent to user@example.com)

Using pattern matching:

>>> def process_result(result):
...     return result.match(
...         left_fn=lambda error: f"Operation failed: {error}",
...         right_fn=lambda value: f"Operation succeeded with value: {value}"
...     )
>>>
>>> success = Either.right(42)
>>> failure = Either.left("Invalid input")
>>>
>>> print(process_result(success))  # Output: Operation succeeded with value: 42
>>> print(process_result(failure))  # Output: Operation failed: Invalid input

class monawhat.either.Either

Bases: Generic[A, B], BaseMonad[B]

Base class for the Either monad. This class should not be instantiated directly.

Either represents a value of one of two possible types (a disjoint union). An instance of Either is either a Left or a Right.

get() → A | B

Return the contained value.

get_or_else(default: B) → B

Return the value if this is a Right, or the default value if it’s a Left.

is_left() → bool

Check if this Either is a Left.

is_right() → bool

Check if this Either is a Right.

static left(value: A) → Left[A, B]

Create a Left instance containing the given value.

match(left_fn: Callable[[A], C], right_fn: Callable[[B], C]) → C

Apply the appropriate function based on whether this is a Left or Right.

Parameters:

• left_fn – Function to apply if this is a Left

• right_fn – Function to apply if this is a Right

Returns:

The result of applying the appropriate function

static right(value: B) → Right[A, B]

Create a Right instance containing the given value.

class monawhat.either.Left(value: A)

Bases: Either[A, B]

The Left case of the Either monad, typically representing failure.

__repr__() → str

Return a string representation of the Left instance.

get() → A

Return the contained value.

get_or_else(default: B) → B

Return the default value.

is_left() → bool

Check if this Either is a Left.

is_right() → bool

Check if this Either is a Right.

class monawhat.either.Right(value: B)

Bases: Either[A, B]

The Right case of the Either monad, typically representing success.

__repr__() → str

Return a string representation of the Right instance.

get() → B

Return the contained value.

get_or_else(default: B) → B

Return the contained value.

is_left() → bool

Check if this Either is a Left.

is_right() → bool

Check if this Either is a Right.

monawhat.identity module

Module implementing the Identity monad, the simplest monad that just wraps a value.

Examples

Create and use an Identity monad:

>>> x = Identity(5)
>>> x.get()
5

Map a function over an Identity:

>>> x = Identity(5)
>>> y = x.map(lambda n: n * 2)
>>> y.get()
10

Chain Identity monads using bind:

>>> x = Identity(5)
>>> y = x.bind(lambda n: Identity(n * 2))
>>> y.get()
10

Create an Identity using pure:

>>> x = Identity.pure(5)
>>> x.get()
5

class monawhat.identity.Identity(value: A)

Bases: Generic[A]

Identity monad that simply wraps a value.

The Identity monad is the simplest monad that just wraps a value and provides monadic operations on it. It serves as a baseline for monadic computations and can be useful as a building block for more complex monads.

__repr__() → str

Return a string representation of the Identity monad.

Returns:

A string representation of the Identity with its wrapped value.

bind(f: Callable[[A], Identity[B]]) → Identity[B]

Chain this Identity with a function that returns another Identity.

Parameters:

f – A function that takes the value of this Identity and returns a new Identity.

Returns:

The Identity returned by the function f.

get() → A

Get the value wrapped by this Identity.

Returns:

The wrapped value.

map(f: Callable[[A], B]) → Identity[B]

Apply a function to the wrapped value.

Parameters:

f – The function to apply to the value.

Returns:

A new Identity containing the result of applying f to the value.

static pure(value: A) → Identity[A]

Create an Identity containing the given value.

Parameters:

value – The value to wrap in an Identity.

Returns:

An Identity containing the value.

monawhat.io module

Module implementing IO monads for handling input/output operations.

Examples

Basic IOLite usage:

>>> from monawhat.io import IOLite
>>> read_name = IOLite.from_callable(input, "Enter name: ")
>>> greet = read_name.map(lambda name: f"Hello, {name}!")
>>> greet.run()  # Executes the IO operation

Full IO with error handling:

>>> from monawhat.io import IO
>>> def divide(x: int, y: int) -> IO[float]:
...     return IO.from_callable(lambda: x / y)
>>> safe_divide = divide(10, 0).attempt()
>>> result = safe_divide.run()  # Returns Either[Exception, float]

IO composition:

>>> get_input = IO.input("Enter text: ")
>>> print_upper = get_input.map(str.upper).bind(IO.print)
>>> print_upper.run()  # Gets input and prints it in uppercase

Error handling with catch:

>>> def safe_operation() -> IO[str]:
...     return IO.from_callable(lambda: some_risky_operation()).catch(
...         lambda e: IO.pure("Operation failed")
...     )

class monawhat.io.IO(effect: Callable[[], A])

Bases: BaseMonad[A]

IO monad for representing and composing IO operations.

The IO monad represents a computation that may perform input/output operations and produce a value. It’s useful for making IO explicit and composable.

Unlike in Haskell, this IO monad doesn’t enforce purity - it’s primarily for composition and making effects more explicit.

__repr__() → str

Return a string representation of the IO.

static async_to_io(coro: Callable[[...], Awaitable[A]], *args: Any, **kwargs: Any) → IO[A]

Convert an async function to an IO.

Parameters:

• coro – An async function

• *args – Arguments to pass to the function

• **kwargs – Keyword arguments to pass to the function

Returns:

An IO that runs the async function

attempt() → IO[Either[Exception, A]]

Convert this IO into one that returns an Either for error handling.

Returns:

An IO that returns Either a Left with the exception or a Right with the result

catch(handler: Callable[[Exception], IO[A]]) → IO[A]

Catch exceptions thrown by this IO and handle them with the provided function.

Parameters:

handler – A function that takes an exception and returns an IO

Returns:

A new IO that catches and handles exceptions

static fail(exception: Exception) → IO[A]

Create an IO that raises an exception when run.

Parameters:

exception – The exception to raise

Returns:

An IO that raises the exception

static for_each(xs: list[A], f: Callable[[A], IO[B]]) → IO[None]

Apply an IO-producing function to each element in a list for its effects.

Parameters:

• xs – A list of values

• f – A function that takes a value and returns an IO

Returns:

An IO that executes each operation for its effect

static from_callable(io_operation: Callable[[...], A], *args: Any, **kwargs: Any) → IO[A]

Create an IO from any callable.

Parameters:

• io_operation – The function to call

• *args – Arguments to pass to the function

• **kwargs – Keyword arguments to pass to the function

Returns:

An IO that calls the function

static input(prompt: str = '') → IO[str]

Create an IO that reads a line from stdin.

Parameters:

prompt – The prompt to display

Returns:

An IO that reads a line

map_error(f: Callable[[Exception], Exception]) → IO[A]

Transform any exception thrown by this IO operation.

Parameters:

f – A function that transforms exceptions

Returns:

A new IO that transforms exceptions

static print(message: str, end: str = '\n', file: TextIO | None = None) → IO[None]

Create an IO that prints a message.

Parameters:

• message – The message to print

• end – The string to append after the message

• file – The file to print to (defaults to sys.stdout)

Returns:

An IO that prints the message

run() → A

Execute the IO operation and return its result.

Returns:

The result of the IO operation

static sequence(ios: list[IO[A]]) → IO[list[A]]

Execute a list of IO operations in sequence and collect their results.

Parameters:

ios – A list of IO operations

Returns:

An IO that returns a list of results

then(io: IO[B]) → IO[B]

Execute this IO, ignore its result, and then execute another IO.

Parameters:

io – The IO operation to execute after this one

Returns:

A new IO that represents the sequential execution

static traverse(xs: list[A], f: Callable[[A], IO[B]]) → IO[list[B]]

Apply an IO-producing function to each element in a list and collect the results.

Parameters:

• xs – A list of values

• f – A function that takes a value and returns an IO

Returns:

An IO that returns a list of results

static with_context_io(ctx_manager: AbstractContextManager[T]) → IO[T]

Create an IO from a context manager.

This allows using Python’s context managers within the IO monad.

Parameters:

ctx_manager – The context manager that yields a value of type T

Returns:

An IO that represents the context manager’s yielded value

with_finally(finalizer: IO[Any]) → IO[A]

Ensure the finalizer runs regardless of whether this IO succeeds or fails.

Parameters:

finalizer – The IO to run after this IO, regardless of outcome

Returns:

A new IO that runs the finalizer after this IO

class monawhat.io.IOLite(effect: Callable[[], A])

Bases: BaseMonad[A]

Lightweight IO monad for representing and composing IO operations.

This is a minimal implementation focused on core monad operations. For a more feature-rich implementation, see the IO class.

__repr__() → str

Return a string representation of the IO.

static from_callable(io_operation: Callable[[...], A], *args: Any, **kwargs: Any) → IOLite[A]

Create an IO from any callable.

run() → A

Execute the IO operation and return its result.

monawhat.io.io_either_bind(io_either: IO[Either[E, A]], f: Callable[[A], IO[Either[E, B]]]) → IO[Either[E, B]]

Bind an IO[Either] with a function that takes the Right value and returns another IO[Either].

This is useful for chaining operations that can fail.

Parameters:

• io_either – An IO that returns an Either

• f – A function that takes the success value and returns an IO[Either]

Returns:

An IO that returns the final Either

monawhat.maybe module

Module implementing the Option monad for handling optional values.

Examples

Create a Just value:

>>> maybe_value = Maybe.just(42)
>>> maybe_value.get()
42

Create a Nothing value:

>>> empty = Maybe.nothing()
>>> empty.is_nothing()
True

Map over a Maybe:

>>> maybe_value = Maybe.just(42)
>>> doubled = maybe_value.map(lambda x: x * 2)
>>> doubled.get()
84

Handle None values:

>>> value = Maybe.from_optional(None)
>>> value.get_or_else(0)
0

Chain computations:

>>> def divide(x: int, y: int) -> Maybe[float]:
...     return Maybe.just(x / y) if y != 0 else Maybe.nothing()
>>> result = Maybe.just(10).bind(lambda x: divide(x, 2))
>>> result.get()
5.0

class monawhat.maybe.Just(value: A)

Bases: Maybe[A]

The Just case of the Option monad, representing a present value.

__repr__() → str

Return a string representation of the Just instance.

get() → A

Return the contained value.

get_or_else(default: A) → A

Return the contained value.

is_just() → bool

Check if this Option is a Just.

is_nothing() → bool

Check if this Option is a Nothing.

class monawhat.maybe.Maybe

Bases: BaseMonad[A]

Base class for the Option monad. This class should not be instantiated directly.

Option represents a value that may or may not be present. An instance of Option is either a Just (containing a value) or a Nothing.

static from_optional(value: A | None) → Maybe[A]

Create a Maybe from an optional value.

Parameters:

value – Value that might be None

Returns:

Just(value) if value is not None, otherwise Nothing()

get() → A

Return the contained value if Just, or raise an error if Nothing.

get_or_else(default: A) → A

Return the value if this is a Just, or the default value if it’s a Nothing.

is_just() → bool

Check if this Option is a Just.

is_nothing() → bool

Check if this Option is a Nothing.

static just(value: A) → Just[A]

Create a Just instance containing the given value.

static nothing() → Nothing[A]

Create a Nothing instance.

to_either(error: E) → Either[E, A]

Convert this Maybe to an Either.

Args: error: The error value to use if this is Nothing

Returns: Either.left(error) if this is Nothing, otherwise Either.right(self.get())

class monawhat.maybe.Nothing

Bases: Maybe[A]

The Nothing case of the Option monad, representing the absence of a value.

__repr__() → str

Return a string representation of the Nothing instance.

get() → A

Raise an error since Nothing contains no value.

get_or_else(default: A) → A

Return the default value.

is_just() → bool

Check if this Option is a Just.

is_nothing() → bool

Check if this Option is a Nothing.

monawhat.reader module

Module implementing the Reader monad for handling environment-based computations.

Examples

Basic usage of Reader monad:

>>> # Create a Reader that reads a config value
>>> get_db_url = Reader.asks(lambda config: config['db_url'])
>>> # Create a Reader that uses the db_url
>>> def get_user(user_id: int) -> Reader:
...     return get_db_url.map(lambda url: f"Fetching user {user_id} from {url}")
>>> # Run the computation with a config
>>> config = {'db_url': 'postgresql://localhost:5432'}
>>> result = get_user(123).run(config)
>>> print(result)
'Fetching user 123 from postgresql://localhost:5432'

Chaining Reader computations:

>>> # Create Readers for different config values
>>> get_host = Reader.asks(lambda config: config['host'])
>>> get_port = Reader.asks(lambda config: config['port'])
>>> # Combine them using bind
>>> def get_address() -> Reader:
...     return get_host.bind(
...         lambda h: get_port.map(
...             lambda p: f"{h}:{p}"
...         )
...     )
>>> # Run with config
>>> config = {'host': 'localhost', 'port': 8080}
>>> result = get_address().run(config)
>>> print(result)
'localhost:8080'

class monawhat.reader.Reader(run_fn: Callable[[E], A])

Bases: Generic[E, A], BaseMonad[A]

Reader monad for computations that read values from a shared environment.

The Reader monad represents a computation that can read values from a shared environment and produce a result. It’s useful for dependency injection and managing configurations.

__repr__() → str

Return a string representation of the Reader.

static ask() → Reader[E, E]

Create a Reader that returns the environment itself.

Returns:

A Reader that returns the environment it’s run with.

static asks(f: Callable[[E], A]) → Reader[E, A]

Create a Reader that applies a function to the environment.

Parameters:

f – A function to apply to the environment.

Returns:

A Reader that applies the function to the environment.

local(f: Callable[[E], E]) → Reader[E, A]

Run this Reader in a modified environment.

Parameters:

f – A function that transforms the environment.

Returns:

A Reader that runs in the modified environment.

run(env: E) → A

Execute the Reader computation with the given environment.

Parameters:

env – The environment to run the computation in.

Returns:

The result of the computation.

monawhat.state module

Module implementing the State monad for handling stateful computations.

Examples

Basic usage of State monad:

>>> # Create a state that increments a counter and returns the old value
>>> def increment() -> State[int, int]:
...     return State(lambda s: (s, s + 1))
...
>>> # Chain multiple state operations
>>> computation = increment().bind(
...     lambda x: increment().bind(
...         lambda y: State.pure(x + y)
...     )
... )
>>> # Run the computation with initial state 0
>>> result, final_state = computation.run(0)
>>> print(result)  # Sum of first two states: 0 + 1 = 1
1
>>> print(final_state)  # Final counter value: 2

Using helper methods:

>>> # Get current state
>>> get_state = State.get()
>>> value, state = get_state.run(42)
>>> assert value == state == 42

>>> # Modify state
>>> add_one = State.modify(lambda x: x + 1)
>>> _, new_state = add_one.run(10)
>>> assert new_state == 11

class monawhat.state.State(run_fn: Callable[[S], tuple[A, S]])

Bases: Generic[S, A], BaseMonad[A]

State monad for computations that manipulate state.

The State monad represents a stateful computation that produces a value and potentially modifies some state. It’s useful for modeling computations with mutable state in a purely functional way.

A State[S, A] instance represents a function from state S to a tuple (A, S), where A is the result value and S is the new state.

__repr__() → str

Return a string representation of the State.

eval(initial_state: S) → A

Execute the computation and return only the result value.

Parameters:

initial_state – The starting state

Returns:

The result value

exec(initial_state: S) → S

Execute the computation and return only the final state.

Parameters:

initial_state – The starting state

Returns:

The final state

static get() → State[S, S]

Create a State that returns the current state as its value.

Returns:

A State that returns the current state as its value

static gets(f: Callable[[S], A]) → State[S, A]

Create a State that applies a function to the current state to produce a value.

Parameters:

f – A function that computes a value from the state

Returns:

A State that applies the function to the current state

static modify(f: Callable[[S], S]) → State[S, None]

Create a State that modifies the current state using a function.

Parameters:

f – A function that transforms the state

Returns:

A State that applies the function to the current state

static put(new_state: S) → State[S, None]

Create a State that replaces the current state.

Parameters:

new_state – The new state to use

Returns:

A State that replaces the current state

run(initial_state: S) → tuple[A, S]

Execute the stateful computation with the given initial state.

Parameters:

initial_state – The starting state

Returns:

A tuple containing the result value and the final state

monawhat.writer module

Module implementing the Writer monad for accumulating computations with output.

Examples

Basic usage with string output:

>>> w1 = Writer(5, "first ")  # Create a Writer with value 5 and output "first "
>>> w2 = w1.map(lambda x: x * 2)  # Map value to 10, output unchanged
>>> w2.run()
(10, "first ")

Chaining Writers with bind:

>>> def add_one(x):
...     return Writer(x + 1, "added one ")
>>> w3 = w1.bind(add_one)  # Chain operations, outputs combine
>>> w3.run()
(6, "first added one ")

Using tell to add output:

>>> w4 = Writer.tell("log message ")  # Create Writer with just output
>>> w4.run()
(None, "log message ")

Using listen to access output:

>>> w5 = Writer(42, "answer ").listen()  # Value becomes tuple with output
>>> w5.run()
((42, "answer "), "answer ")

class monawhat.writer.Monoid(*args, **kwargs)

Bases: Protocol

Protocol defining a type that supports the + operator (like a monoid).

__add__(other: Self) → Self

Combine two monoid values.

class monawhat.writer.Writer(value: A, output: W, combine: Callable[[W, W], W] | None = None)

Bases: Generic[W, A], BaseMonad[A]

Writer monad for computations that produce a value along with accumulated output.

The Writer monad represents a computation that produces a result along with some accumulated output (like a log). It’s useful for tracking computations that generate additional data alongside their primary result.

The output type W should form a monoid, meaning it has: - An associative binary operation for combining values (+ operator) - An identity element that, when combined with any value, gives that value back

Common examples: lists, strings, numbers with addition, etc.

__repr__() → str

Return a string representation of the Writer.

censor(f: Callable[[W], W]) → Writer[W, A]

Apply a function to the output while preserving the value.

Parameters:

f – The function to apply to the output.

Returns:

A new Writer with the same value and transformed output.

listen() → Writer[W, tuple[A, W]]

Create a Writer where the value includes the output as well.

Returns:

A new Writer where the value is paired with the output.

output() → W

Get the output from the Writer.

Returns:

The accumulated output.

pass_output() → Writer[W, A]

Execute this Writer assuming its value is a function that transforms output.

The function in value should have the signature Callable[[W], W].

Returns:

A Writer where the output is transformed by the function in the value.

classmethod pure(value: A, empty: W, combine: Callable[[W, W], W] | None = None) → Writer[W, A]

Create a Writer with a value and empty output.

Parameters:

• value – The value to wrap.

• empty – The identity element for the output monoid.

• combine – Optional function to combine outputs. If None, uses the + operator.

Returns:

A Writer containing the value and empty output.

run() → tuple[A, W]

Extract the value and output from the Writer.

Returns:

A tuple containing the result value and accumulated output.

classmethod tell(output: W, combine: Callable[[W, W], W] | None = None) → Writer[W, None]

Create a Writer that only produces output with no meaningful value.

Parameters:

• output – The output to produce.

• combine – Optional function to combine outputs. If None, uses the + operator.

Returns:

A Writer with the given output and None as the value.

value() → A

Get the value from the Writer.

Returns:

The result value.

Module contents