FunML

A collection of utilities to help write python as though it were an ML-kind of functional language like OCaml.

Provides:

1. Immutable data structures like enums, records, lists
2. Piping outputs of one function to another as inputs. That's how bigger functions are created from smaller ones.
3. Pattern matching for declarative conditional control of flow instead of using 'if's
4. Error handling using the Result monad, courtesy of rust. Instead of using try-except all over the place, functions return a Result which has the right data when successful and an exception if unsuccessful. The result is then pattern-matched to retrieve the data or react to the exception.
5. No None. Instead, we use the Option monad, courtesy of rust. When an Option has data, it is Option.SOME, or else it is Option.NONE. Pattern matching helps handle both scenarios.

Enum

Bases: types.MLType

Enumerable type that can only be in a limited number of forms.

Other enums are created by inheriting from this type. An enum can only be in a limited number of forms or variant and each variant can have some data associated with each instance.

Variants are created by setting class attributes. The value of the class attributes should be the shape of the associated data or None if variant has no associated data.

The pre-created types of Option and Result are both enums

Raises:

Type	Description
TypeError	got unexpected data type, different from the signature

Example

import funml as ml
from datetime import date

class Day(ml.Enum):
    MON = date
    TUE = date
    WED = date
    THUR = date
    FRI = date
    SAT = date
    SUN = date

dates =  [
    date(200, 3, 4),
    date(2009, 1, 16),
    date(1993, 12, 29),
    date(2004, 10, 13),
    date(2020, 9, 5),
    date(2004, 5, 7),
    date(1228, 8, 18),
]

to_day_enum = lambda date_value: (
    ml.match(date_value.weekday())
        .case(0, do=lambda: Day.MON(date_value))
        .case(1, do=lambda: Day.TUE(date_value))
        .case(2, do=lambda: Day.WED(date_value))
        .case(3, do=lambda: Day.THUR(date_value))
        .case(4, do=lambda: Day.FRI(date_value))
        .case(5, do=lambda: Day.SAT(date_value))
        .case(6, do=lambda: Day.SUN(date_value))
)()

day_enums_transform = ml.imap(to_day_enum)
day_enums = day_enums_transform(dates)

print(day_enums)
# prints [<Day.TUE: datetime.date(200, 3, 4)>, <Day.FRI: datetime.date(2009, 1, 16)>,        # <Day.WED: datetime.date(1993, 12, 29)>, <Day.WED: datetime.date(2004, 10, 13)>,         # <Day.SAT: datetime.date(2020, 9, 5)>, <Day.FRI: datetime.date(2004, 5, 7)>,         # <Day.FRI: datetime.date(1228, 8, 18)>]

Source code in funml/data/enum.py

class Enum(types.MLType):
    """Enumerable type that can only be in a limited number of forms.

    Other enums are created by inheriting from this type.
    An enum can only be in a limited number of forms or variant and each variant
    can have some data associated with each instance.

    Variants are created by setting class attributes. The value of the class
    attributes should be the shape of the associated data or `None` if variant has no
    associated data.

    The pre-created types of [`Option`][funml.Option] and [`Result`][funml.Result]
    are both enums

    Raises:
        TypeError: got unexpected data type, different from the signature

    Example:
        ```python
        import funml as ml
        from datetime import date

        class Day(ml.Enum):
            MON = date
            TUE = date
            WED = date
            THUR = date
            FRI = date
            SAT = date
            SUN = date

        dates =  [
            date(200, 3, 4),
            date(2009, 1, 16),
            date(1993, 12, 29),
            date(2004, 10, 13),
            date(2020, 9, 5),
            date(2004, 5, 7),
            date(1228, 8, 18),
        ]

        to_day_enum = lambda date_value: (
            ml.match(date_value.weekday())
                .case(0, do=lambda: Day.MON(date_value))
                .case(1, do=lambda: Day.TUE(date_value))
                .case(2, do=lambda: Day.WED(date_value))
                .case(3, do=lambda: Day.THUR(date_value))
                .case(4, do=lambda: Day.FRI(date_value))
                .case(5, do=lambda: Day.SAT(date_value))
                .case(6, do=lambda: Day.SUN(date_value))
        )()

        day_enums_transform = ml.imap(to_day_enum)
        day_enums = day_enums_transform(dates)

        print(day_enums)
        # prints [<Day.TUE: datetime.date(200, 3, 4)>, <Day.FRI: datetime.date(2009, 1, 16)>,\
        # <Day.WED: datetime.date(1993, 12, 29)>, <Day.WED: datetime.date(2004, 10, 13)>, \
        # <Day.SAT: datetime.date(2020, 9, 5)>, <Day.FRI: datetime.date(2004, 5, 7)>, \
        # <Day.FRI: datetime.date(1228, 8, 18)>]
        ```
    """

    signature: Optional[Union[Tuple[Type, ...], Dict[str, Type], Type]] = None
    _name: str = ""

    def __init_subclass__(cls, **kwargs):
        """Creating a new Enum"""
        slots = []

        for k, v in _get_cls_attrs(cls).items():
            cls.__add_variant(k, v)
            slots.append(k)

        cls.__slots__ = slots

    def __init__(self, *args: Union[Any, Dict[str, Any], int]):
        if isinstance(self.signature, tuple):
            data = args
        else:
            data = args[0]

        if not _is_valid(data, self.signature):
            raise TypeError(
                f"expected data type passed to be {self.signature}, got {data} from args: {args}"
            )

        self._value = data

    @classmethod
    def __add_variant(
        cls,
        name: str,
        shape: Optional[Union[Type, Tuple[Type, ...], Dict[str, Type]]] = None,
    ) -> Type["Enum"]:
        """Adds a given option to the enum.

        This is a chainable method that can be used to add multiple options
        to the same enum class.

        Args:
            name: the name of the option e.g. `ERR` for the [`Result`][funml.Result] enum
            shape: the signature of the associated data

        Returns:
             the Enum class to which the option has been attached.
        """
        dotted_name = f"{cls.__name__}.{name}"
        signature = shape

        if signature is None:
            opt_value = cls(name)
        else:
            # each option is a subclass of this enum class
            opt_value = type(dotted_name, (cls,), {})

        opt_value._name = dotted_name
        opt_value.signature = signature

        setattr(cls, name, opt_value)
        return cls

    @property
    def value(self):
        """The value associated with this Enum option, if any"""
        return self._value

    @property
    def name(self):
        """The name of this Enum"""
        return self._name

    def generate_case(self, do: types.Operation) -> Tuple[Callable, types.Expression]:
        """See Base Class: [`MLType`][funml.types.MLType]"""
        op = lambda *args: do(*args)
        if self._value is not None:
            op = lambda arg: do(_get_enum_captured_value(arg))

        return self._is_like, types.Expression(types.Operation(func=op))

    def _is_like(self, other) -> bool:
        """See Base Class: [`MLType`][funml.types.MLType]"""
        if not isinstance(other, Enum):
            return False

        return (
            self.__class__ == other.__class__
            and self._name == other._name
            and (self._value == other._value or _is_valid(other._value, self._value))
        )

    def __eq__(self, other: "Enum"):
        """Checks equality of the this enum and `other`.

        Args:
            other: the value to compare with current enum.
        """
        return (
            self.__class__ == other.__class__
            and self._name == other._name
            and utils.equals(self._value, other._value)
        )

    def __str__(self):
        """Generates a readable presentation of the enum."""
        return f"<{self.name}: {self.value}>"

    def __repr__(self):
        return f"<{self.name}: {self.value}>"

name property

The name of this Enum

value property

The value associated with this Enum option, if any

__add_variant(name, shape=None) classmethod

Adds a given option to the enum.

This is a chainable method that can be used to add multiple options to the same enum class.

Parameters:

Name	Type	Description	Default
name	str	the name of the option e.g. ERR for the Result enum	required
shape	Optional[Union[Type, Tuple[Type, ...], Dict[str, Type]]]	the signature of the associated data	None

Returns:

Type	Description
Type[Enum]	the Enum class to which the option has been attached.

Source code in funml/data/enum.py

@classmethod
def __add_variant(
    cls,
    name: str,
    shape: Optional[Union[Type, Tuple[Type, ...], Dict[str, Type]]] = None,
) -> Type["Enum"]:
    """Adds a given option to the enum.

    This is a chainable method that can be used to add multiple options
    to the same enum class.

    Args:
        name: the name of the option e.g. `ERR` for the [`Result`][funml.Result] enum
        shape: the signature of the associated data

    Returns:
         the Enum class to which the option has been attached.
    """
    dotted_name = f"{cls.__name__}.{name}"
    signature = shape

    if signature is None:
        opt_value = cls(name)
    else:
        # each option is a subclass of this enum class
        opt_value = type(dotted_name, (cls,), {})

    opt_value._name = dotted_name
    opt_value.signature = signature

    setattr(cls, name, opt_value)
    return cls

__eq__(other)

Checks equality of the this enum and other.

Parameters:

Name	Type	Description	Default
other	Enum	the value to compare with current enum.	required

Source code in funml/data/enum.py

def __eq__(self, other: "Enum"):
    """Checks equality of the this enum and `other`.

    Args:
        other: the value to compare with current enum.
    """
    return (
        self.__class__ == other.__class__
        and self._name == other._name
        and utils.equals(self._value, other._value)
    )

__init_subclass__(**kwargs)

Creating a new Enum

Source code in funml/data/enum.py

def __init_subclass__(cls, **kwargs):
    """Creating a new Enum"""
    slots = []

    for k, v in _get_cls_attrs(cls).items():
        cls.__add_variant(k, v)
        slots.append(k)

    cls.__slots__ = slots

__str__()

Generates a readable presentation of the enum.

Source code in funml/data/enum.py

def __str__(self):
    """Generates a readable presentation of the enum."""
    return f"<{self.name}: {self.value}>"

_is_like(other)

See Base Class: MLType

Source code in funml/data/enum.py

def _is_like(self, other) -> bool:
    """See Base Class: [`MLType`][funml.types.MLType]"""
    if not isinstance(other, Enum):
        return False

    return (
        self.__class__ == other.__class__
        and self._name == other._name
        and (self._value == other._value or _is_valid(other._value, self._value))
    )

generate_case(do)

See Base Class: MLType

Source code in funml/data/enum.py

def generate_case(self, do: types.Operation) -> Tuple[Callable, types.Expression]:
    """See Base Class: [`MLType`][funml.types.MLType]"""
    op = lambda *args: do(*args)
    if self._value is not None:
        op = lambda arg: do(_get_enum_captured_value(arg))

    return self._is_like, types.Expression(types.Operation(func=op))

IList

Bases: types.MLType, Generic[T]

An immutable list of items of any type.

Parameters:

Name	Type	Description	Default
args	T	the items to be included in the list.	()

Source code in funml/data/lists.py

class IList(types.MLType, Generic[T]):
    """An immutable list of items of any type.

    Args:
        args: the items to be included in the list.
    """

    def __init__(self, *args: T):
        self.__size: Optional[int] = None
        self.__capture_start: Optional[int] = None
        self.__capture_tail_len: int = 0
        self.__pre_capture: Optional[List[T]] = None
        self.__post_capture: Optional[List[T]] = None
        self.__list: Optional[List[T]] = None
        self._head: Optional["_Node"] = None

        args = _eval_generator_args(args)
        self.__set_size_from_args(args)
        self.__initialize_from_tuple(args)

    @property
    def head(self) -> T:
        """The first item in the list."""
        return self._head.value

    @property
    def tail(self) -> "IList[T]":
        """A new slice of the list containing all items except the first."""
        return IList.__from_node(self._head.next)

    def generate_case(self, do: types.Operation):
        """See Base class: [`MLType`][funml.types.MLType]"""
        start = 0 if self.__capture_start is None else self.__capture_start
        tail_len = self.__capture_tail_len

        def op(arg):
            arg_slice = arg[start : (len(arg) - tail_len)]
            return do(arg_slice)

        return self._is_like, types.Expression(types.Operation(func=op))

    def _is_like(self, other: Any) -> bool:
        """See Base Class: [`MLType`][funml.types.MLType]"""
        if not isinstance(other, IList):
            return False

        if self._size > other._size:
            return False

        if self.__capture_start is None:
            return self == other

        pre_capture = other._self_list[: self.__capture_start]
        post_capture = other._self_list[(len(other) - self.__capture_tail_len) :]

        return _lists_match(
            schema=self._pre_capture, actual=pre_capture
        ) and _lists_match(schema=self._post_capture, actual=post_capture)

    @property
    def _size(self) -> int:
        """The number of items in the list."""
        if self.__size is None:
            self.__size = len(self._self_list)
        return self.__size

    @property
    def _self_list(self) -> List[T]:
        """A cache of the native list that corresponds to this list."""
        if self.__list is None:
            self.__list = list(self.__iter__())
        return self.__list

    @property
    def _pre_capture(self) -> List[T]:
        """A slice of the list pattern before the section to be captured when matching."""
        if self.__pre_capture is None and self.__capture_start is not None:
            self.__pre_capture = self._self_list[: self.__capture_start]
        return self.__pre_capture

    @property
    def _post_capture(self) -> List[T]:
        """A slice of the list pattern after the section to be captured when matching."""
        if self.__post_capture is None:
            self.__post_capture = self._self_list[
                (self._size - self.__capture_tail_len) :
            ]
        return self.__post_capture

    @classmethod
    def __from_node(cls, head: "_Node[T]") -> "IList[T]":
        """Generates a slice of the old IList given one node of that list.

        In this case, the new list shares the same memory as the old list
        so don't use this in scenarios where immutable lists are needed.

        Args:
            head: the node from which the new list is to start from.

        Returns:
            A new list that shares memory with the old list. **NOTE: This is not immutable. Don't use it**.
        """
        i_list = IList()
        i_list._head = head
        return i_list

    def __set_size_from_args(self, args: Tuple[T]):
        """Updates the size of this list basing on the args passed.

        Args:
            args: the items to be put in this list.
        """
        args_len = len(args)
        if args_len > 0:
            self.__size = args_len

    def __initialize_from_tuple(self, args: Tuple[T]):
        """Initializes the list using items passed to it as a tuple.

        Initializes the current IList, generating nodes corresponding to the args passed
        and setting any capture sections if `...` is found.

        Args:
            args: the items to include in the list
        """
        prev: Optional[_Node] = None
        for i, v in enumerate(reversed(args)):
            node = _Node(_data=v, _next=prev)
            prev = node

            if v is ...:
                self.__capture_start = self._size - i - 1
                self.__capture_tail_len = i

        self._head: Optional["_Node"] = prev

    def __len__(self):
        """Computes the length of the list."""
        return self._size

    def __iter__(self):
        """Makes the list an iterable."""
        if self._head is None:
            return

        yield self._head.value

        curr = self._head.next
        while curr is not None:
            yield curr.value
            curr = curr.next

    def __add__(self, other: "IList[T]") -> "IList[T]":
        """Creates a new list with the current list and the `other` list merged.

        Args:
            other: the list to be appended to current list when creating new merged list.

        Returns:
            A new list which is a combination of the current list and the `other` list.

        Raises:
            TypeError: other is not an `IList`
        """
        if not isinstance(other, IList):
            raise TypeError(
                f"add operation requires value to be of type IList, not {type(other)}"
            )

        return IList(*self, *other)

    def __getitem__(self, item: Union[slice, int]) -> Union["IList[T]", Any]:
        """Makes this list subscriptable and sliceable.

        Args:
            item: the index or slice to return.

        Returns:
            An `IList` if `index` was a slice or an item in the list if index was an integer.

        Raises:
            IndexError: if `item` is out of range of the list.
        """
        if isinstance(item, slice):
            return IList(*self._self_list[item])
        return self._self_list[item]

    def __eq__(self, other: Any) -> bool:
        """Checks equality of the this list and `other`.

        Args:
            other: the value to compare with current list.
        """
        return utils.equals(self._self_list, other._self_list)

    def __str__(self):
        """Generates a readable presentation of the list."""
        map_to_str = imap(str)
        return f"[{', '.join(map_to_str(self))}]"

    def __repr__(self):
        map_to_str = imap(str)
        return f"IList({', '.join(map_to_str(self))})"

_post_capture: List[T] property

A slice of the list pattern after the section to be captured when matching.

_pre_capture: List[T] property

A slice of the list pattern before the section to be captured when matching.

_self_list: List[T] property

A cache of the native list that corresponds to this list.

_size: int property

The number of items in the list.

head: T property

The first item in the list.

tail: IList[T] property

A new slice of the list containing all items except the first.

__add__(other)

Creates a new list with the current list and the other list merged.

Parameters:

Name	Type	Description	Default
other	IList[T]	the list to be appended to current list when creating new merged list.	required

Returns:

Type	Description
IList[T]	A new list which is a combination of the current list and the other list.

Raises:

Type	Description
TypeError	other is not an IList

Source code in funml/data/lists.py

def __add__(self, other: "IList[T]") -> "IList[T]":
    """Creates a new list with the current list and the `other` list merged.

    Args:
        other: the list to be appended to current list when creating new merged list.

    Returns:
        A new list which is a combination of the current list and the `other` list.

    Raises:
        TypeError: other is not an `IList`
    """
    if not isinstance(other, IList):
        raise TypeError(
            f"add operation requires value to be of type IList, not {type(other)}"
        )

    return IList(*self, *other)

__eq__(other)

Checks equality of the this list and other.

Parameters:

Name	Type	Description	Default
other	Any	the value to compare with current list.	required

Source code in funml/data/lists.py

def __eq__(self, other: Any) -> bool:
    """Checks equality of the this list and `other`.

    Args:
        other: the value to compare with current list.
    """
    return utils.equals(self._self_list, other._self_list)

__from_node(head) classmethod

Generates a slice of the old IList given one node of that list.

In this case, the new list shares the same memory as the old list so don't use this in scenarios where immutable lists are needed.

Parameters:

Name	Type	Description	Default
head	_Node[T]	the node from which the new list is to start from.	required

Returns:

Type	Description
IList[T]	A new list that shares memory with the old list. NOTE: This is not immutable. Don't use it.

Source code in funml/data/lists.py

@classmethod
def __from_node(cls, head: "_Node[T]") -> "IList[T]":
    """Generates a slice of the old IList given one node of that list.

    In this case, the new list shares the same memory as the old list
    so don't use this in scenarios where immutable lists are needed.

    Args:
        head: the node from which the new list is to start from.

    Returns:
        A new list that shares memory with the old list. **NOTE: This is not immutable. Don't use it**.
    """
    i_list = IList()
    i_list._head = head
    return i_list

__getitem__(item)

Makes this list subscriptable and sliceable.

Parameters:

Name	Type	Description	Default
item	Union[slice, int]	the index or slice to return.	required

Returns:

Type	Description
Union[IList[T], Any]	An IList if index was a slice or an item in the list if index was an integer.

Raises:

Type	Description
IndexError	if item is out of range of the list.

Source code in funml/data/lists.py

def __getitem__(self, item: Union[slice, int]) -> Union["IList[T]", Any]:
    """Makes this list subscriptable and sliceable.

    Args:
        item: the index or slice to return.

    Returns:
        An `IList` if `index` was a slice or an item in the list if index was an integer.

    Raises:
        IndexError: if `item` is out of range of the list.
    """
    if isinstance(item, slice):
        return IList(*self._self_list[item])
    return self._self_list[item]

__initialize_from_tuple(args)

Initializes the list using items passed to it as a tuple.

Initializes the current IList, generating nodes corresponding to the args passed and setting any capture sections if ... is found.

Parameters:

Name	Type	Description	Default
args	Tuple[T]	the items to include in the list	required

Source code in funml/data/lists.py

def __initialize_from_tuple(self, args: Tuple[T]):
    """Initializes the list using items passed to it as a tuple.

    Initializes the current IList, generating nodes corresponding to the args passed
    and setting any capture sections if `...` is found.

    Args:
        args: the items to include in the list
    """
    prev: Optional[_Node] = None
    for i, v in enumerate(reversed(args)):
        node = _Node(_data=v, _next=prev)
        prev = node

        if v is ...:
            self.__capture_start = self._size - i - 1
            self.__capture_tail_len = i

    self._head: Optional["_Node"] = prev

__iter__()

Makes the list an iterable.

Source code in funml/data/lists.py

def __iter__(self):
    """Makes the list an iterable."""
    if self._head is None:
        return

    yield self._head.value

    curr = self._head.next
    while curr is not None:
        yield curr.value
        curr = curr.next

__len__()

Computes the length of the list.

Source code in funml/data/lists.py

def __len__(self):
    """Computes the length of the list."""
    return self._size

__set_size_from_args(args)

Updates the size of this list basing on the args passed.

Parameters:

Name	Type	Description	Default
args	Tuple[T]	the items to be put in this list.	required

Source code in funml/data/lists.py

def __set_size_from_args(self, args: Tuple[T]):
    """Updates the size of this list basing on the args passed.

    Args:
        args: the items to be put in this list.
    """
    args_len = len(args)
    if args_len > 0:
        self.__size = args_len

__str__()

Generates a readable presentation of the list.

Source code in funml/data/lists.py

def __str__(self):
    """Generates a readable presentation of the list."""
    map_to_str = imap(str)
    return f"[{', '.join(map_to_str(self))}]"

_is_like(other)

See Base Class: MLType

Source code in funml/data/lists.py

def _is_like(self, other: Any) -> bool:
    """See Base Class: [`MLType`][funml.types.MLType]"""
    if not isinstance(other, IList):
        return False

    if self._size > other._size:
        return False

    if self.__capture_start is None:
        return self == other

    pre_capture = other._self_list[: self.__capture_start]
    post_capture = other._self_list[(len(other) - self.__capture_tail_len) :]

    return _lists_match(
        schema=self._pre_capture, actual=pre_capture
    ) and _lists_match(schema=self._post_capture, actual=post_capture)

generate_case(do)

See Base class: MLType

Source code in funml/data/lists.py

def generate_case(self, do: types.Operation):
    """See Base class: [`MLType`][funml.types.MLType]"""
    start = 0 if self.__capture_start is None else self.__capture_start
    tail_len = self.__capture_tail_len

    def op(arg):
        arg_slice = arg[start : (len(arg) - tail_len)]
        return do(arg_slice)

    return self._is_like, types.Expression(types.Operation(func=op))

Option

Bases: Enum

Represents a value that is potentially None

Variants

• SOME: when an actual value exists
• NONE: when there is no value

Example Usage

import funml as ml
from typing import Any

b = ml.Option.SOME(6)
a = ml.Option.NONE
extract_option = (ml.match()
        .case(ml.Option.SOME(Any), do=lambda v: v)
        .case(ml.Option.NONE, do=lambda: "nothing found"))
extract_option(b)
# returns 6
extract_option(a)
# returns 'nothing found'

Source code in funml/data/monads.py

class Option(Enum):
    """Represents a value that is potentially None

    Variants:
        - SOME: when an actual value exists
        - NONE: when there is no value

    Example Usage:

        ```python
        import funml as ml
        from typing import Any

        b = ml.Option.SOME(6)
        a = ml.Option.NONE
        extract_option = (ml.match()
                .case(ml.Option.SOME(Any), do=lambda v: v)
                .case(ml.Option.NONE, do=lambda: "nothing found"))
        extract_option(b)
        # returns 6
        extract_option(a)
        # returns 'nothing found'
        ```
    """

    NONE = None
    SOME = Any

Result

Bases: Enum

Represents a value that is potentially an exception

Variants

• ERR: when an exception is raised
• OK: when there is a real value

Example

import funml as ml
from typing import Any

b = ml.Result.OK(60)
a = ml.Result.ERR(TypeError("some error"))
extract_result = (ml.match()
        .case(ml.Result.OK(Any), do=lambda v: v)
        .case(ml.Result.ERR(Exception), do=lambda v: str(v)))
extract_result(b)
# returns 60
extract_result(a)
# returns 'some error'

Source code in funml/data/monads.py

class Result(Enum):
    """Represents a value that is potentially an exception

    Variants:
        - ERR: when an exception is raised
        - OK: when there is a real value

    Example:

        ```python
        import funml as ml
        from typing import Any

        b = ml.Result.OK(60)
        a = ml.Result.ERR(TypeError("some error"))
        extract_result = (ml.match()
                .case(ml.Result.OK(Any), do=lambda v: v)
                .case(ml.Result.ERR(Exception), do=lambda v: str(v)))
        extract_result(b)
        # returns 60
        extract_result(a)
        # returns 'some error'
        ```
    """

    ERR = Exception
    OK = Any

execute(*args, **kwargs)

Executes a pipeline returning its output.

A pipeline will be executed the moment this expression is reached.

Don't use >> after a call to execute as the pipeline would have already terminated.

Parameters:

Name	Type	Description	Default
args	Any	any arguments to run on the pipeline	()
kwargs	Any	any key-word arguments to run on the pipeline.	{}

Example

import funml as ml

to_power_of = ml.val(lambda power, v: v**power)
divided_by = ml.val(lambda divisor, v: v / divisor)

output = ml.val(90) >> to_power_of(3) >> divided_by(90) >> divided_by(3) >> ml.execute()
# prints 2700

Source code in funml/pipeline.py

def execute(*args: Any, **kwargs: Any) -> ExecutionExpression:
    """Executes a pipeline returning its output.

    A pipeline will be executed the moment this expression is
    reached.

    Don't use `>>` after a call to execute as the pipeline
    would have already terminated.

    Args:
        args: any arguments to run on the pipeline
        kwargs: any key-word arguments to run on the pipeline.

    Example:
        ```python
        import funml as ml

        to_power_of = ml.val(lambda power, v: v**power)
        divided_by = ml.val(lambda divisor, v: v / divisor)

        output = ml.val(90) >> to_power_of(3) >> divided_by(90) >> divided_by(3) >> ml.execute()
        # prints 2700
        ```
    """
    return ExecutionExpression(*args, **kwargs)

from_json(type_, value, strict=True)

Converts a JSON string into the given type.

If strict is True, an error is returned if the JSON string cannot be converted into the type, else if strict is False and an error occurs, the default python primitive is str, dict etc. is returned.

Parameters:

Name	Type	Description	Default
type_	Type[T]	the typing annotation to which the JSON string is to be converted to	required
value	str	the JSON string	required
strict	bool	whether the JSON string should be strictly converted to the given type or left as the default primitive python objects	True

Returns:

Type	Description
T	the instance got from the JSON string

Raises:

Type	Description
ValueError	unable to deserialize JSON to given type

Example

import funml as ml


@ml.record
class Student:
    name: str
    favorite_color: "Color"

@ml.record
class Color:
    r: int
    g: int
    b: int
    a: "Alpha"

class Alpha(ml.Enum):
    OPAQUE = None
    TRANSLUCENT = float

items = [
    (
        ml.IList[Color],
        (
        "["
        '{"name": "John Doe", "favorite_color": {"r": 8, "g": 4, "b": 78, "a": "Alpha.OPAQUE: \"OPAQUE\""}}, '
        '{"name": "Jane Doe", "favorite_color": {"r": 55, "g": 40, "b": 9, "a": "Alpha.TRANSLUCENT: 0.4"}}'
        "]"
        )
    ),
    (Color, '{"r": 55, "g": 40, "b": 9, "a": "Alpha.TRANSLUCENT: 0.4"}'),
    (Alpha, "Alpha.TRANSLUCENT: 0.4"),
]

# setting strict to False can allow any json string to be converted to a python object,
# first attempting to convert it to the provided type, and if it fails,
# the output of an ordinary json.loads call is returned.
#
# However, when strict is True, ValueError's will be raised
# if the json string can't be converted into the given type
strict = True

for type_, item_json in items:
    item = ml.from_json(type_=type_, value=item_json, strict=strict)
    print(item)

Source code in funml/json.py

def from_json(type_: Type[T], value: str, strict: bool = True) -> T:
    """Converts a JSON string into the given type.

    If strict is True, an error is returned if the JSON string cannot be converted into
    the type, else if strict is False and an error occurs,
    the default python primitive is str, dict etc.
    is returned.

    Args:
        type_: the typing annotation to which the JSON string is to be converted to
        value: the JSON string
        strict: whether the JSON string should be strictly converted to the given type
            or left as the default primitive python objects

    Returns:
        the instance got from the JSON string

    Raises:
        ValueError: unable to deserialize JSON to given type

    Example:
        ```python
        import funml as ml


        @ml.record
        class Student:
            name: str
            favorite_color: "Color"

        @ml.record
        class Color:
            r: int
            g: int
            b: int
            a: "Alpha"

        class Alpha(ml.Enum):
            OPAQUE = None
            TRANSLUCENT = float

        items = [
            (
                ml.IList[Color],
                (
                "["
                '{"name": "John Doe", "favorite_color": {"r": 8, "g": 4, "b": 78, "a": "Alpha.OPAQUE: \\"OPAQUE\\""}}, '
                '{"name": "Jane Doe", "favorite_color": {"r": 55, "g": 40, "b": 9, "a": "Alpha.TRANSLUCENT: 0.4"}}'
                "]"
                )
            ),
            (Color, '{"r": 55, "g": 40, "b": 9, "a": "Alpha.TRANSLUCENT: 0.4"}'),
            (Alpha, "Alpha.TRANSLUCENT: 0.4"),
        ]

        # setting strict to False can allow any json string to be converted to a python object,
        # first attempting to convert it to the provided type, and if it fails,
        # the output of an ordinary json.loads call is returned.
        #
        # However, when strict is True, ValueError's will be raised
        # if the json string can't be converted into the given type
        strict = True

        for type_, item_json in items:
            item = ml.from_json(type_=type_, value=item_json, strict=strict)
            print(item)
        ```
    """
    actual_type = extract_type(type_)
    frame = inspect.currentframe()

    try:
        _globals = frame.f_back.f_globals
        _locals = frame.f_back.f_locals

        if issubclass(actual_type, Enum):
            return _enum_from_json(
                actual_type, value, strict, _globals=_globals, _locals=_locals
            )
        if issubclass(actual_type, Record):
            return _record_from_json(
                actual_type, value, strict, _globals=_globals, _locals=_locals
            )
        if issubclass(actual_type, IList):
            return _i_list_from_json(
                type_, value, strict, _globals=_globals, _locals=_locals
            )

        obj = json.loads(value)
        if strict:
            return _cast_to_annotation(
                type_, value=obj, _globals=_globals, _locals=_locals
            )
        else:
            return _try_cast_to_annotation(
                type_, value=obj, _globals=_globals, _locals=_locals
            )
    finally:
        del frame

if_err(do, strict=True)

Does the given operation if value passed to resulting expression is Result.ERR.

If the value is Result.OK, it just returns the Result.OK without doing anything about it.

Parameters:

Name	Type	Description	Default
do	Union[Expression, Callable, Any]	The expression, function, to run or value to return when Result.ERR	required
strict	bool	if only Results should be expected	True

Example

import funml as ml

ok_value = ml.Result.OK(90)
err_value = ml.Result.ERR(TypeError("some stuff"))
another_value = None

# in case the value may not be a Result, set strict to False
print(ml.if_err(str, strict=False)(another_value))
# prints None

err_to_str = ml.if_err(str)
print(err_to_str(err_value))
# prints 'some stuff'

print(err_to_str(ok_value))
# prints <Result.OK: ('90',)>

Returns:

Type	Description
Expression	An expression to run the do operation when value passed to expression is Result.ERR
Expression	or to just return the Result.OK

Raises:

Type	Description
funml.errors.MatchError	value provided was not a Result if strict is True

Source code in funml/data/monads.py

def if_err(do: Union[Expression, Callable, Any], strict: bool = True) -> Expression:
    """Does the given operation if value passed to resulting expression is Result.ERR.

    If the value is Result.OK, it just returns the Result.OK without
    doing anything about it.

    Args:
        do: The expression, function, to run or value to return when Result.ERR
        strict: if only Results should be expected

    Example:
        ```python
        import funml as ml

        ok_value = ml.Result.OK(90)
        err_value = ml.Result.ERR(TypeError("some stuff"))
        another_value = None

        # in case the value may not be a Result, set strict to False
        print(ml.if_err(str, strict=False)(another_value))
        # prints None

        err_to_str = ml.if_err(str)
        print(err_to_str(err_value))
        # prints 'some stuff'

        print(err_to_str(ok_value))
        # prints <Result.OK: ('90',)>
        ```

    Returns:
        An expression to run the `do` operation when value passed to expression is Result.ERR
        or to just return the Result.OK

    Raises:
        funml.errors.MatchError: value provided was not a Result if strict is True
    """
    routine = (
        match()
        .case(Result.OK(Any), do=lambda v: Result.OK(v))
        .case(Result.ERR(Exception), do=to_expn(do))
    )

    if not strict:
        routine = routine.case(Any, do=lambda v: v)

    return Expression(Operation(routine))

if_none(do, strict=True)

Does the given operation if value passed to resulting expression is Option.NONE.

If the value is Option.SOME, it just returns the Option.SOME without doing anything about it.

Parameters:

Name	Type	Description	Default
do	Union[Expression, Callable, Any]	The expression, function, to run or value to return when Option.NONE	required
strict	bool	if only Options should be expected	True

Example

import funml as ml

some_value = ml.Option.SOME(90)
none_value = ml.Option.NONE
another_value = None

# in case the value may not be an Option, set strict to False
print(ml.if_none(str, strict=False)(another_value))
# prints None

none_to_str = ml.if_none(str)
print(none_to_str(some_value))
# prints <Option.SOME: (90,)>

print(none_to_str(none_value))
# prints ('NONE',)

Returns:

Type	Description
Expression	An expression to run the do operation when value passed to expression is Option.NONE
Expression	or to just return the Option.SOME

Raises:

Type	Description
funml.errors.MatchError	value provided was not an Option if strict is True

Source code in funml/data/monads.py

def if_none(do: Union[Expression, Callable, Any], strict: bool = True) -> Expression:
    """Does the given operation if value passed to resulting expression is Option.NONE.

    If the value is Option.SOME, it just returns the Option.SOME without
    doing anything about it.

    Args:
        do: The expression, function, to run or value to return when Option.NONE
        strict: if only Options should be expected

    Example:
        ```python
        import funml as ml

        some_value = ml.Option.SOME(90)
        none_value = ml.Option.NONE
        another_value = None

        # in case the value may not be an Option, set strict to False
        print(ml.if_none(str, strict=False)(another_value))
        # prints None

        none_to_str = ml.if_none(str)
        print(none_to_str(some_value))
        # prints <Option.SOME: (90,)>

        print(none_to_str(none_value))
        # prints ('NONE',)
        ```

    Returns:
        An expression to run the `do` operation when value passed to expression is Option.NONE
        or to just return the Option.SOME

    Raises:
        funml.errors.MatchError: value provided was not an Option if strict is True
    """
    routine = (
        match()
        .case(Option.SOME(Any), do=lambda v: Option.SOME(v))
        .case(Option.NONE, do=to_expn(do))
    )

    if not strict:
        routine = routine.case(Any, do=lambda v: v)

    return Expression(Operation(routine))

if_ok(do, strict=True)

Does the given operation if value passed to resulting expression is Result.OK.

If the value is Result.ERR, it just returns the Result.ERR without doing anything about it.

Parameters:

Name	Type	Description	Default
do	Union[Expression, Callable, Any]	The expression, function to run or value to return when Result.OK	required
strict	bool	if only Results should be expected	True

Example

import funml as ml

ok_value = ml.Result.OK(90)
err_value = ml.Result.ERR(TypeError("some stuff"))
another_value = None

# in case the value may not be a Result, set strict to False
print(ml.if_ok(str, strict=False)(another_value))
# prints None

ok_to_str = ml.if_ok(str)
print(ok_to_str(err_value))
# prints <Result.ERR: (TypeError('some stuff'),)>

print(ok_to_str(ok_value))
# prints 90

Returns:

Type	Description
Expression	An expression to run the do operation when value passed to expression is Result.OK
Expression	or to just return the Result.ERR

Raises:

Type	Description
funml.errors.MatchError	value provided was not a Result and strict is True

Source code in funml/data/monads.py

def if_ok(do: Union[Expression, Callable, Any], strict: bool = True) -> Expression:
    """Does the given operation if value passed to resulting expression is Result.OK.

    If the value is Result.ERR, it just returns the Result.ERR without
    doing anything about it.

    Args:
        do: The expression, function to run or value to return when Result.OK
        strict: if only Results should be expected

    Example:
        ```python
        import funml as ml

        ok_value = ml.Result.OK(90)
        err_value = ml.Result.ERR(TypeError("some stuff"))
        another_value = None

        # in case the value may not be a Result, set strict to False
        print(ml.if_ok(str, strict=False)(another_value))
        # prints None

        ok_to_str = ml.if_ok(str)
        print(ok_to_str(err_value))
        # prints <Result.ERR: (TypeError('some stuff'),)>

        print(ok_to_str(ok_value))
        # prints 90
        ```

    Returns:
        An expression to run the `do` operation when value passed to expression is Result.OK
        or to just return the Result.ERR

    Raises:
        funml.errors.MatchError: value provided was not a Result and `strict` is True
    """
    routine = (
        match()
        .case(Result.OK(Any), do=to_expn(do))
        .case(Result.ERR(Exception), do=lambda v: Result.ERR(v))
    )

    if not strict:
        routine = routine.case(Any, do=lambda v: v)

    return Expression(Operation(routine))

if_some(do, strict=True)

Does the given operation if value passed to resulting expression is Option.SOME.

If the value is Result.NONE, it just returns the Result.NONE without doing anything about it.

Parameters:

Name	Type	Description	Default
do	Union[Expression, Callable, Any]	The expression, function, to run or value to return when Option.SOME	required
strict	bool	if only Options should be expected	True

Example

import funml as ml

some_value = ml.Option.SOME(90)
none_value = ml.Option.NONE
another_value = None

# in case the value may not be an Option, set strict to False
print(ml.if_some(str, strict=False)(another_value))
# prints None

some_to_str = ml.if_some(str)
print(some_to_str(some_value))
# prints 90

print(some_to_str(none_value))
# prints <Option.NONE: ('NONE',)>

Returns:

Type	Description
Expression	An expression to run the do operation when value passed to expression is Option.SOME
Expression	or to just return the Option.NONE

Raises:

Type	Description
funml.errors.MatchError	value provided was not an Option if strict is True

Source code in funml/data/monads.py

def if_some(do: Union[Expression, Callable, Any], strict: bool = True) -> Expression:
    """Does the given operation if value passed to resulting expression is Option.SOME.

    If the value is Result.NONE, it just returns the Result.NONE without
    doing anything about it.

    Args:
        do: The expression, function, to run or value to return when Option.SOME
        strict: if only Options should be expected

    Example:
        ```python
        import funml as ml

        some_value = ml.Option.SOME(90)
        none_value = ml.Option.NONE
        another_value = None

        # in case the value may not be an Option, set strict to False
        print(ml.if_some(str, strict=False)(another_value))
        # prints None

        some_to_str = ml.if_some(str)
        print(some_to_str(some_value))
        # prints 90

        print(some_to_str(none_value))
        # prints <Option.NONE: ('NONE',)>
        ```

    Returns:
        An expression to run the `do` operation when value passed to expression is Option.SOME
        or to just return the Option.NONE

    Raises:
        funml.errors.MatchError: value provided was not an Option if strict is True
    """
    routine = (
        match()
        .case(Option.SOME(Any), do=to_expn(do))
        .case(Option.NONE, do=lambda: Option.NONE)
    )

    if not strict:
        routine = routine.case(Any, do=lambda v: v)

    return Expression(Operation(routine))

ifilter(func)

Creates an expression to transform each item by the given function.

Expressions can be computed lazily at any time.

Parameters:

Name	Type	Description	Default
func	Callable[[Any], Any]	the function to use to check if item should remain or be ignored.	required

Returns:

Type	Description
Expression	A new iList with only the items of data that returned true when func was called on them.

Source code in funml/data/lists.py

def ifilter(func: Callable[[Any], Any]) -> Expression:
    """Creates an expression to transform each item by the given function.

    Expressions can be computed lazily at any time.

    Args:
        func: the function to use to check if item should remain or be ignored.

    Returns:
        A new iList with only the items of data that returned true when `func` was called on them.
    """
    op = Operation(lambda data: IList(*filter(func, data)))
    return Expression(op)

imap(func)

Creates an expression to transform each item by the given function.

Expressions can be computed lazily at any time.

Parameters:

Name	Type	Description	Default
func	Callable[[Any], Any]	the function to use to transform each item	required

Returns:

Type	Description
Expression	A new IList with each item in data transformed according to the func function.

Source code in funml/data/lists.py

def imap(func: Callable[[Any], Any]) -> Expression:
    """Creates an expression to transform each item by the given function.

    Expressions can be computed lazily at any time.

    Args:
        func: the function to use to transform each item

    Returns:
            A new IList with each item in data transformed according to the `func` function.
    """
    op = Operation(lambda data: IList(*[func(v) for v in data]))
    return Expression(op)

ireduce(func, initial=None)

Creates an expression to reduce a sequence into one value using the given func.

Expressions can be computed lazily at any time.

Parameters:

Name	Type	Description	Default
func	Callable[[Any, Any], Any]	the function to reduce the sequence into a single value.	required
initial	Optional[Any]	the initial value that acts like a default when sequence is empty, and is added onto by func when sequence has some items.	None

Returns:

Type	Description
Expression	A single item got from calling func repeatedly across the sequence for every
Expression	two adjacent items.

Source code in funml/data/lists.py

def ireduce(
    func: Callable[[Any, Any], Any], initial: Optional[Any] = None
) -> Expression:
    """Creates an expression to reduce a sequence into one value using the given `func`.

    Expressions can be computed lazily at any time.

    Args:
        func: the function to reduce the sequence into a single value.
        initial: the initial value that acts like a default when sequence is empty,
                and is added onto by `func` when sequence has some items.

    Returns:
        A single item got from calling `func` repeatedly across the sequence for every
        two adjacent items.
    """
    if initial is None:
        op = Operation(lambda data: reduce(func, data))
    else:
        op = Operation(lambda data: reduce(func, data, initial))
    return Expression(op)

is_err(v, strict=True)

Checks if v is Result.ERR.

Parameters:

Name	Type	Description	Default
v	Result	The value to check	required
strict	bool	if value should be a Result	True

Example

import funml as ml

ok_value = ml.Result.OK(90)
err_value = ml.Result.ERR(TypeError())

print(ml.is_err(ok_value))
# prints False

print(ml.is_err(err_value))
# prints True

another_value = None

# in case the value is not always a Result, set `strict` to False
# to avoid a MatchError
print(ml.is_err(another_value, strict=False))
# prints False

Returns:

Type	Description
bool	True if v is a Result.ERR else False

Raises:

Type	Description
funml.errors.MatchError	value provided was not a Result if strict is True

Source code in funml/data/monads.py

def is_err(v: "Result", strict: bool = True) -> bool:
    """Checks if `v` is Result.ERR.

    Args:
        v: The value to check
        strict: if value should be a Result

    Example:
        ```python
        import funml as ml

        ok_value = ml.Result.OK(90)
        err_value = ml.Result.ERR(TypeError())

        print(ml.is_err(ok_value))
        # prints False

        print(ml.is_err(err_value))
        # prints True

        another_value = None

        # in case the value is not always a Result, set `strict` to False
        # to avoid a MatchError
        print(ml.is_err(another_value, strict=False))
        # prints False
        ```

    Returns:
        True if `v` is a Result.ERR else False

    Raises:
        funml.errors.MatchError: value provided was not a Result if strict is True
    """
    negative_pattern = Result.OK(Any) if strict else Any

    return (
        match()
        .case(Result.ERR(Exception), do=lambda: True)
        .case(negative_pattern, do=lambda: False)
    )(v)

is_none(v, strict=True)

Checks if v is Option.NONE.

Parameters:

Name	Type	Description	Default
v	Option	The value to check	required
strict	bool	if value should be a Option	True

Example

import funml as ml

some_value = ml.Option.SOME(90)
none_value = ml.Option.NONE

print(ml.is_none(some_value))
# prints False

print(ml.is_none(none_value))
# prints True

another_value = None

# in case the value is not always an Option, set `strict` to False
# to avoid a MatchError
print(ml.is_none(another_value, strict=False))
# prints False

Returns:

Type	Description
bool	True if v is a Option.NONE else False

Raises:

Type	Description
funml.errors.MatchError	value provided was not an Option if strict is True

Source code in funml/data/monads.py

def is_none(v: "Option", strict: bool = True) -> bool:
    """Checks if `v` is Option.NONE.

    Args:
        v: The value to check
        strict: if value should be a Option

    Example:
        ```python
        import funml as ml

        some_value = ml.Option.SOME(90)
        none_value = ml.Option.NONE

        print(ml.is_none(some_value))
        # prints False

        print(ml.is_none(none_value))
        # prints True

        another_value = None

        # in case the value is not always an Option, set `strict` to False
        # to avoid a MatchError
        print(ml.is_none(another_value, strict=False))
        # prints False
        ```

    Returns:
        True if `v` is a Option.NONE else False

    Raises:
        funml.errors.MatchError: value provided was not an Option if strict is True
    """
    negative_pattern = Option.SOME(Any) if strict else Any

    return (
        match()
        .case(Option.NONE, do=lambda: True)
        .case(negative_pattern, do=lambda: False)
    )(v)

is_ok(v, strict=True)

Checks if v is Result.OK.

Parameters:

Name	Type	Description	Default
v	Result	The value to check	required
strict	bool	if value should be a Result	True

Example

import funml as ml

ok_value = ml.Result.OK(90)
err_value = ml.Result.ERR(TypeError())

print(ml.is_ok(ok_value))
# prints True

print(ml.is_ok(err_value))
# prints False

another_value = None

# in case the value is not always a Result, set `strict` to False
# to avoid a MatchError
print(ml.is_ok(another_value, strict=False))
# prints False

Returns:

Type	Description
bool	True if v is a Result.OK else False

Raises:

Type	Description
funml.errors.MatchError	value provided was not a Result if strict is True

Source code in funml/data/monads.py

def is_ok(v: "Result", strict: bool = True) -> bool:
    """Checks if `v` is Result.OK.

    Args:
        v: The value to check
        strict: if value should be a Result

    Example:
        ```python
        import funml as ml

        ok_value = ml.Result.OK(90)
        err_value = ml.Result.ERR(TypeError())

        print(ml.is_ok(ok_value))
        # prints True

        print(ml.is_ok(err_value))
        # prints False

        another_value = None

        # in case the value is not always a Result, set `strict` to False
        # to avoid a MatchError
        print(ml.is_ok(another_value, strict=False))
        # prints False
        ```

    Returns:
        True if `v` is a Result.OK else False

    Raises:
        funml.errors.MatchError: value provided was not a Result if strict is True
    """
    negative_pattern = Result.ERR(Exception) if strict else Any

    return (
        match()
        .case(Result.OK(Any), do=lambda: True)
        .case(negative_pattern, do=lambda: False)
    )(v)

is_some(v, strict=True)

Checks if v is Option.SOME.

Parameters:

Name	Type	Description	Default
v	Option	The value to check	required
strict	bool	if value should be a Option	True

Example

import funml as ml

some_value = ml.Option.SOME(90)
none_value = ml.Option.NONE

print(ml.is_some(some_value))
# prints True

print(ml.is_some(none_value))
# prints False

another_value = None

# in case the value is not always an Option, set `strict` to False
# to avoid a MatchError
print(ml.is_some(another_value, strict=False))
# prints False

Returns:

Type	Description
bool	True if v is a Option.SOME else False

Raises:

Type	Description
funml.errors.MatchError	value provided was not an Option if strict is True

Source code in funml/data/monads.py

def is_some(v: "Option", strict: bool = True) -> bool:
    """Checks if `v` is Option.SOME.

    Args:
        v: The value to check
        strict: if value should be a Option

    Example:
        ```python
        import funml as ml

        some_value = ml.Option.SOME(90)
        none_value = ml.Option.NONE

        print(ml.is_some(some_value))
        # prints True

        print(ml.is_some(none_value))
        # prints False

        another_value = None

        # in case the value is not always an Option, set `strict` to False
        # to avoid a MatchError
        print(ml.is_some(another_value, strict=False))
        # prints False
        ```

    Returns:
        True if `v` is a Option.SOME else False

    Raises:
        funml.errors.MatchError: value provided was not an Option if strict is True
    """
    negative_pattern = Option.NONE if strict else Any

    return (
        match()
        .case(Option.SOME(Any), do=lambda: True)
        .case(negative_pattern, do=lambda: False)
    )(v)

l(*args)

Creates an immutable list of any type of items.

Creates a list of items of any type, that cannot be changed once created. It can only be used to create other lists, using methods on it like

• + - to combine two separate lists into a new one containing elements of both
• imap(fn) - to create a new list with each element transformed according to the given function fn
• filter(fn) - to return a new list containing only elements that conform to the given function fn

Parameters:

Name	Type	Description	Default
args	Any	the items that make up the list	()

Returns:

Type	Description
IList	An immutable list, `IList, containing the items passed to it.

Example

import funml as ml

items = ml.l(120, 13, 40, 60, "hey", "men")
# or items = ml.l(item for item in values) where values is an iterable

num_filter = ml.ifilter(lambda x: isinstance(x, (int, float)))
str_filter = ml.ifilter(lambda x: isinstance(x, str))
nums = num_filter(items)
strings = str_filter(items)

double_transform = ml.imap(lambda x: x*2)
doubled_nums = double_transform(nums)

aggregator = ml.ireduce(lambda x, y: f"{x}, {y}")
list_as_str = aggregator(items)

print(nums)
# prints [120, 13, 40, 60]

print(strings)
# prints ["hey", "men"]

print(doubled_nums)
# prints [240, 26, 80, 120]

print(list_as_str)
# prints '120, 13, 40, 60, hey, men'

Source code in funml/data/lists.py

def l(*args: Any) -> "IList":
    """Creates an immutable list of any type of items.

    Creates a list of items of any type, that cannot be changed
    once created. It can only be used to create other lists, using methods on it like

    - [`+`][funml.IList.__add__] - to combine two separate lists into a new one containing elements of both
    - [`imap(fn)`][funml.imap] - to create a new list with each element transformed according to the given function `fn`
    - [`filter(fn)`][funml.ifilter] - to return a new list containing only elements that conform to the given function `fn`

    Args:
        args: the items that make up the list

    Returns:
        An immutable list, [`IList][funml.IList], containing the items passed to it.

    Example:
        ```python
        import funml as ml

        items = ml.l(120, 13, 40, 60, "hey", "men")
        # or items = ml.l(item for item in values) where values is an iterable

        num_filter = ml.ifilter(lambda x: isinstance(x, (int, float)))
        str_filter = ml.ifilter(lambda x: isinstance(x, str))
        nums = num_filter(items)
        strings = str_filter(items)

        double_transform = ml.imap(lambda x: x*2)
        doubled_nums = double_transform(nums)

        aggregator = ml.ireduce(lambda x, y: f"{x}, {y}")
        list_as_str = aggregator(items)

        print(nums)
        # prints [120, 13, 40, 60]

        print(strings)
        # prints ["hey", "men"]

        print(doubled_nums)
        # prints [240, 26, 80, 120]

        print(list_as_str)
        # prints '120, 13, 40, 60, hey, men'
        ```
    """
    return IList(*args)

match(arg=None)

Matches the argument with a corresponding case, and calls it do operation.

It runs through a range of cases, looking for any that match the current arg. If it finds it, it calls the do operation attached to that case.

Parameters:

Name	Type	Description	Default
arg	Optional[Any]	the argument which is to be checked against the list of patterns.	None

Returns:

Type	Description
MatchExpression	A MatchExpression containing the matched operation.

Raises:

Type	Description
MatchError	failed to find a match for the argument.

Example

import funml as ml

@ml.record
class Color:
    r: int
    g: int
    b: int

raw_value = ml.Option.SOME(90)
value = (
    ml.match(raw_value)
        .case(ml.Option.SOME(int), do=lambda v: v + 6)
        .case(ml.Option.NONE, do=lambda: "nothing to show")
        .case(Color(red=255), do=lambda v: v.red + v.green)
        .case(ml.l(..., 5), do=lambda v: v)
        .case(ml.l(8, 5, ...), do=lambda v: str(v))
)()
print(value)
# prints 96

Source code in funml/pattern_match.py

def match(arg: Optional[Any] = None) -> "MatchExpression":
    """Matches the argument with a corresponding case, and calls it `do` operation.

    It runs through a range of cases, looking for any that match
    the current arg. If it finds it, it calls the `do` operation attached
    to that case.

    Args:
        arg: the argument which is to be checked against the list of patterns.

    Returns:
        A [`MatchExpression`][funml.types.MatchExpression] containing the matched operation.

    Raises:
        MatchError: failed to find a match for the argument.

    Example:

        ```python
        import funml as ml

        @ml.record
        class Color:
            r: int
            g: int
            b: int

        raw_value = ml.Option.SOME(90)
        value = (
            ml.match(raw_value)
                .case(ml.Option.SOME(int), do=lambda v: v + 6)
                .case(ml.Option.NONE, do=lambda: "nothing to show")
                .case(Color(red=255), do=lambda v: v.red + v.green)
                .case(ml.l(..., 5), do=lambda v: v)
                .case(ml.l(8, 5, ...), do=lambda v: str(v))
        )()
        print(value)
        # prints 96
        ```
    """
    return MatchExpression(arg=arg)

record(cls)

Creates a Schema type to create similar records.

Used usually as a decorator inplace of @dataclass on dataclass-like classes to make them ml-functional.

It creates a Schema for similar objects that contain a given set of attributes. For example, it can create a Book schema whose attributes include author, title, isbn etc.

Parameters:

Name	Type	Description	Default
cls	Type[R]	the class to transform into a record	required

Returns:

Type	Description
Type[R]	A class which can act as a record of the particular schema set by the attributes.

Example

import funml as ml

@ml.record
class Color:
    red: int
    green: int
    blue: int
    alpha: int = 1

indigo = Color(red=75, green=0, blue=130)

print(indigo)
# prints {'red': 75, 'green': 0, 'blue': 130, 'alpha': 1}

Source code in funml/data/records.py

@dataclass_transform(
    field_specifiers=(dataclasses.Field, dataclasses.field),
)
def record(cls: Type[R]) -> Type[R]:
    """Creates a Schema type to create similar records.

    Used usually as a decorator inplace of @dataclass
    on dataclass-like classes to make them ml-functional.

    It creates a Schema for similar objects that contain a given
    set of attributes. For example, it can create a `Book` schema
    whose attributes include `author`, `title`, `isbn` etc.

    Args:
        cls: the class to transform into a record

    Returns:
        A class which can act as a record of the particular schema \
        set by the attributes.

    Example:
        ```python
        import funml as ml

        @ml.record
        class Color:
            red: int
            green: int
            blue: int
            alpha: int = 1

        indigo = Color(red=75, green=0, blue=130)

        print(indigo)
        # prints {'red': 75, 'green': 0, 'blue': 130, 'alpha': 1}
        ```
    """
    _annotations = cls.__annotations__

    return dataclasses.dataclass(
        type(
            cls.__name__,
            (
                Record,
                cls,
            ),
            {
                "__annotations__": _annotations,
                "__slots__": tuple(_annotations.keys()),
                "__defaults__": _get_cls_defaults(cls, _annotations=_annotations),
                "__is_normalized__": False,
                "__module_path__": cls.__module__,
            },
        ),
        init=False,
        repr=False,
    )

to_dict(v)

Converts a record into a dictionary.

Parameters:

Name	Type	Description	Default
v	Record	the Record to convert to dict	required

Returns:

Type	Description
Dict[str, Any]	the dictionary representation of the record

Example

import funml as ml


@ml.record
class Department:
    seniors: list[str]
    juniors: List[str]
    locations: tuple[str, ...]
    misc: dict[str, Any]
    head: str

sc_dept = Department(
    seniors=["Joe", "Jane"],
    juniors=["Herbert", "Leo"],
    locations=("Kasasa", "Bujumbura", "Bugahya"),
    misc={"short_name": "ScDept"},
    head="John",
)

data = ml.to_dict(sc_dept)
print(data)

Source code in funml/data/records.py

def to_dict(v: "Record") -> Dict[str, Any]:
    """Converts a record into a dictionary.

    Args:
        v: the `Record` to convert to dict

    Returns:
        the dictionary representation of the record

    Example:
        ```python
        import funml as ml


        @ml.record
        class Department:
            seniors: list[str]
            juniors: List[str]
            locations: tuple[str, ...]
            misc: dict[str, Any]
            head: str

        sc_dept = Department(
            seniors=["Joe", "Jane"],
            juniors=["Herbert", "Leo"],
            locations=("Kasasa", "Bujumbura", "Bugahya"),
            misc={"short_name": "ScDept"},
            head="John",
        )

        data = ml.to_dict(sc_dept)
        print(data)
        ```
    """
    return dict(v)

to_json(value)

Converts the type into a JSON string.

Parameters:

Name	Type	Description	Default
value	Any	the value to convert to a JSON string	required

Returns:

Type	Description
str	the JSON string representation of this instance

Example

import funml as ml


@ml.record
class Student:
    name: str
    favorite_color: "Color"

@ml.record
class Color:
    r: int
    g: int
    b: int
    a: "Alpha"

class Alpha(ml.Enum):
    OPAQUE = None
    TRANSLUCENT = float

items = [
    ml.l(
        True,
        Color(r=8, g=4, b=78, a=Alpha.OPAQUE),
        Color(r=55, g=40, b=9, a=Alpha.TRANSLUCENT(0.4)),
    ),
    Color(r=8, g=4, b=78, a=Alpha.OPAQUE),
    Alpha.TRANSLUCENT(0.4),
]

for item in items:
    item_json = ml.to_json(item)
    print(item_json)

Source code in funml/json.py

def to_json(value: Any) -> str:
    """Converts the type into a JSON string.

    Args:
        value: the value to convert to a JSON string

    Returns:
        the JSON string representation of this instance

    Example:
        ```python
        import funml as ml


        @ml.record
        class Student:
            name: str
            favorite_color: "Color"

        @ml.record
        class Color:
            r: int
            g: int
            b: int
            a: "Alpha"

        class Alpha(ml.Enum):
            OPAQUE = None
            TRANSLUCENT = float

        items = [
            ml.l(
                True,
                Color(r=8, g=4, b=78, a=Alpha.OPAQUE),
                Color(r=55, g=40, b=9, a=Alpha.TRANSLUCENT(0.4)),
            ),
            Color(r=8, g=4, b=78, a=Alpha.OPAQUE),
            Alpha.TRANSLUCENT(0.4),
        ]

        for item in items:
            item_json = ml.to_json(item)
            print(item_json)
        ```
    """
    if isinstance(value, Enum):
        return _enum_to_json(value)
    if isinstance(value, Record):
        return _record_to_json(value)
    if isinstance(value, IList):
        return _i_list_to_json(value)
    if isinstance(value, (tuple, list, set)):
        return f"[{', '.join([to_json(v) for v in value])}]"
    if isinstance(value, dict):
        items = [f'"{k}": {to_json(v)}' for k, v in value.items()]
        return f"{{{', '.join(items)}}}"
    return json.dumps(value)

val(v)

Converts a generic value or lambda expression into a functional expression.

This is useful when one needs to use piping on a non-ml function or value. It is like the connection that give non-ml values and functions capabilities to be used in the ml world.

Parameters:

Name	Type	Description	Default
v	Union[Expression, Callable, Any]	the value e.g. 90 or function e.g. min	required

Returns:

Type	Description
Expression	an ml Expression that can be piped to other ml-expressions or invoked to return its output.

Example

import funml as ml

ml_min = ml.val(min)
ml_min_str = ml_min >> str

expn = ml.val([6, 7, 12]) >> ml_min_str
expn()
# returns '6'

Source code in funml/expressions.py

def val(v: Union[Expression, Callable, Any]) -> Expression:
    """Converts a generic value or lambda expression into a functional expression.

    This is useful when one needs to use piping on a non-ml function or
    value. It is like the connection that give non-ml values and functions
    capabilities to be used in the ml world.

    Args:
        v: the value e.g. 90 or function e.g. `min`

    Returns:
        an ml [`Expression`][funml.types.Expression] that can be piped to other ml-expressions or invoked to return its\
        output.

    Example:

        ```python
        import funml as ml

        ml_min = ml.val(min)
        ml_min_str = ml_min >> str

        expn = ml.val([6, 7, 12]) >> ml_min_str
        expn()
        # returns '6'
        ```
    """
    return to_expn(v)

Types

All types used by funml

Pipeline

A series of logic blocks that operate on the same data in sequence.

This has internal state so it is not be used in such stuff as recursion. However when compile is run on it, a reusable (pure) expression is created.

Source code in funml/types.py

class Pipeline:
    """A series of logic blocks that operate on the same data in sequence.

    This has internal state so it is not be used in such stuff as recursion.
    However when compile is run on it, a reusable (pure) expression is created.
    """

    def __init__(self):
        self._queue: List[Expression] = []
        self._is_terminated = False

    def __rshift__(
        self,
        nxt: Union[
            "Expression",
            Callable,
            "Pipeline",
        ],
    ) -> Union["Pipeline", Any]:
        """Uses `>>` to append the nxt expression, callable, pipeline to this pipeline.

        Args:
            nxt: the next expression, pipeline, or callable to apply after the current one.

        Returns:
            the updated pipeline or the value when the pipeline is executed in case `nxt` is of \
            type `ExecutionExpression`

        Raises:
            ValueError: when the pipeline is already terminated with ml.execute() in its queue.
        """
        self.__update_queue(nxt)
        if self._is_terminated:
            return self()

        return self

    def __call__(
        self, *args: Any, **kwargs: Any
    ) -> Union["Expression", Awaitable["Expression"], Awaitable[Any], Any,]:
        """Computes the logic within the pipeline and returns the value.

        This method runs all those expressions in the queue sequentially,
        with the output of an expression being used as
        input for the next expression.

        Args:
            args: any arguments passed.
            kwargs: any key-word arguments passed

        Returns:
            the computed output of this pipeline or a partial expression if the args and kwargs provided are less than those expected.
        """
        output = None
        queue = self._queue[:-1] if self._is_terminated else self._queue
        is_async = False

        for expn in queue:
            if output is None:
                output = expn(*args, **kwargs)
            elif isinstance(output, Awaitable):
                is_async = True
                break
            else:
                # make sure piped expressions only consume previous outputs args, and kwargs
                output = expn(output, **kwargs)

        if is_async:
            return self.__as_async()(*args, **kwargs)

        return output

    def __copy__(self):
        """Helps call copy on a pipeline"""
        new_pipeline = Pipeline()
        new_pipeline._queue += self._queue
        new_pipeline._is_terminated = self._is_terminated
        return new_pipeline

    def __as_async(self) -> "AsyncPipeline":
        """Creates an async pipeline from this pipeline"""
        pipe = AsyncPipeline()
        pipe._queue = [*self._queue]
        pipe._is_terminated = self._is_terminated
        return pipe

    def __update_queue(
        self,
        nxt: Union[
            "Expression",
            Callable,
            "Pipeline",
        ],
    ):
        """Appends a pipeline or an expression to the queue."""
        if self._is_terminated:
            raise ValueError("a terminated pipeline cannot be extended.")

        if isinstance(nxt, Pipeline):
            self._queue += nxt._queue
            self._is_terminated = nxt._is_terminated
        else:
            nxt_expn = to_expn(nxt)
            self._queue.append(nxt_expn)
            self._is_terminated = isinstance(nxt, ExecutionExpression)

__as_async()

Creates an async pipeline from this pipeline

Source code in funml/types.py

def __as_async(self) -> "AsyncPipeline":
    """Creates an async pipeline from this pipeline"""
    pipe = AsyncPipeline()
    pipe._queue = [*self._queue]
    pipe._is_terminated = self._is_terminated
    return pipe

__call__(*args, **kwargs)

Computes the logic within the pipeline and returns the value.

This method runs all those expressions in the queue sequentially, with the output of an expression being used as input for the next expression.

Parameters:

Name	Type	Description	Default
args	Any	any arguments passed.	()
kwargs	Any	any key-word arguments passed	{}

Returns:

Type	Description
Union[Expression, Awaitable[Expression], Awaitable[Any], Any]	the computed output of this pipeline or a partial expression if the args and kwargs provided are less than those expected.

Source code in funml/types.py

def __call__(
    self, *args: Any, **kwargs: Any
) -> Union["Expression", Awaitable["Expression"], Awaitable[Any], Any,]:
    """Computes the logic within the pipeline and returns the value.

    This method runs all those expressions in the queue sequentially,
    with the output of an expression being used as
    input for the next expression.

    Args:
        args: any arguments passed.
        kwargs: any key-word arguments passed

    Returns:
        the computed output of this pipeline or a partial expression if the args and kwargs provided are less than those expected.
    """
    output = None
    queue = self._queue[:-1] if self._is_terminated else self._queue
    is_async = False

    for expn in queue:
        if output is None:
            output = expn(*args, **kwargs)
        elif isinstance(output, Awaitable):
            is_async = True
            break
        else:
            # make sure piped expressions only consume previous outputs args, and kwargs
            output = expn(output, **kwargs)

    if is_async:
        return self.__as_async()(*args, **kwargs)

    return output

__copy__()

Helps call copy on a pipeline

Source code in funml/types.py

def __copy__(self):
    """Helps call copy on a pipeline"""
    new_pipeline = Pipeline()
    new_pipeline._queue += self._queue
    new_pipeline._is_terminated = self._is_terminated
    return new_pipeline

__rshift__(nxt)

Uses >> to append the nxt expression, callable, pipeline to this pipeline.

Parameters:

Name	Type	Description	Default
nxt	Union[Expression, Callable, Pipeline]	the next expression, pipeline, or callable to apply after the current one.	required

Returns:

Type	Description
Union[Pipeline, Any]	the updated pipeline or the value when the pipeline is executed in case nxt is of type ExecutionExpression

Raises:

Type	Description
ValueError	when the pipeline is already terminated with ml.execute() in its queue.

Source code in funml/types.py

def __rshift__(
    self,
    nxt: Union[
        "Expression",
        Callable,
        "Pipeline",
    ],
) -> Union["Pipeline", Any]:
    """Uses `>>` to append the nxt expression, callable, pipeline to this pipeline.

    Args:
        nxt: the next expression, pipeline, or callable to apply after the current one.

    Returns:
        the updated pipeline or the value when the pipeline is executed in case `nxt` is of \
        type `ExecutionExpression`

    Raises:
        ValueError: when the pipeline is already terminated with ml.execute() in its queue.
    """
    self.__update_queue(nxt)
    if self._is_terminated:
        return self()

    return self

__update_queue(nxt)

Appends a pipeline or an expression to the queue.

Source code in funml/types.py

def __update_queue(
    self,
    nxt: Union[
        "Expression",
        Callable,
        "Pipeline",
    ],
):
    """Appends a pipeline or an expression to the queue."""
    if self._is_terminated:
        raise ValueError("a terminated pipeline cannot be extended.")

    if isinstance(nxt, Pipeline):
        self._queue += nxt._queue
        self._is_terminated = nxt._is_terminated
    else:
        nxt_expn = to_expn(nxt)
        self._queue.append(nxt_expn)
        self._is_terminated = isinstance(nxt, ExecutionExpression)

AsyncPipeline

Bases: Pipeline

A pipeline for handling async code.

See more details in the base class

Source code in funml/types.py

class AsyncPipeline(Pipeline):
    """A pipeline for handling async code.

    See more details in the [base class](funml.types.Pipeline)
    """

    async def __call__(self, *args: Any, **kwargs: Any) -> Union["Expression", Any]:
        """Computes the logic within the pipeline and returns the value.

        This method runs all those expressions in the queue sequentially,
        with the output of an expression being used as
        input for the next expression.

        Args:
            args: any arguments passed.
            kwargs: any key-word arguments passed

        Returns:
            the computed output of this pipeline or a partial expression if the args and kwargs provided \
            are less than expected.
        """
        output = None
        queue = self._queue[:-1] if self._is_terminated else self._queue

        for expn in queue:
            if output is None:
                output = expn(*args, **kwargs)
            elif isinstance(output, Awaitable):
                output = expn((await output), **kwargs)
            else:
                output = expn(output, **kwargs)

        return output

__call__(*args, **kwargs) async

Computes the logic within the pipeline and returns the value.

This method runs all those expressions in the queue sequentially, with the output of an expression being used as input for the next expression.

Parameters:

Name	Type	Description	Default
args	Any	any arguments passed.	()
kwargs	Any	any key-word arguments passed	{}

Returns:

Type	Description
Union[Expression, Any]	the computed output of this pipeline or a partial expression if the args and kwargs provided are less than expected.

Source code in funml/types.py

async def __call__(self, *args: Any, **kwargs: Any) -> Union["Expression", Any]:
    """Computes the logic within the pipeline and returns the value.

    This method runs all those expressions in the queue sequentially,
    with the output of an expression being used as
    input for the next expression.

    Args:
        args: any arguments passed.
        kwargs: any key-word arguments passed

    Returns:
        the computed output of this pipeline or a partial expression if the args and kwargs provided \
        are less than expected.
    """
    output = None
    queue = self._queue[:-1] if self._is_terminated else self._queue

    for expn in queue:
        if output is None:
            output = expn(*args, **kwargs)
        elif isinstance(output, Awaitable):
            output = expn((await output), **kwargs)
        else:
            output = expn(output, **kwargs)

    return output

Expression

Logic that returns a value when applied.

This is the basic building block of all functions and thus almost everything in FunML is converted into an expression at one point or another.

Parameters:

Name	Type	Description	Default
f	Optional[Operation]	the operation or logic to run as part of this expression	None

Source code in funml/types.py

class Expression:
    """Logic that returns a value when applied.

    This is the basic building block of all functions and thus
    almost everything in FunML is converted into an expression at one point or another.

    Args:
        f: the operation or logic to run as part of this expression
    """

    def __init__(self, f: Optional["Operation"] = None):
        self._f = f if f is not None else Operation(lambda x, *args, **kwargs: x)

    def __call__(self, *args: Any, **kwargs: Any) -> Union["Expression", Any]:
        """Computes the logic within and returns the value.

        Args:
            args: any arguments passed.
            kwargs: any key-word arguments passed

        Returns:
            the computed output of this expression or another expression with a partial operation \
            in it if the args provided are less than expected.
        """
        value = self._f(*args, **kwargs)
        if isinstance(value, Operation):
            return Expression(f=value)
        return value

    def __rshift__(
        self,
        nxt: Union[
            "Expression",
            "Pipeline",
            Callable,
        ],
    ) -> Union["Pipeline", Any]:
        """This makes piping using the '>>' symbol possible.

        Combines with the given `nxt` expression or pipeline to produce a new pipeline
        where data flows from current to nxt.

        Args:
            nxt: the next expression, pipeline, or callable to apply after the current one.

        Returns:
            a new pipeline  where the first expression is the current expression followed by `nxt` \
            or returns the value when the pipeline is executed in case `nxt` is of type `ExecutionExpression`
        """
        new_pipeline = Pipeline()
        new_pipeline >> self >> nxt
        return new_pipeline

__call__(*args, **kwargs)

Computes the logic within and returns the value.

Parameters:

Name	Type	Description	Default
args	Any	any arguments passed.	()
kwargs	Any	any key-word arguments passed	{}

Returns:

Type	Description
Union[Expression, Any]	the computed output of this expression or another expression with a partial operation in it if the args provided are less than expected.

Source code in funml/types.py

def __call__(self, *args: Any, **kwargs: Any) -> Union["Expression", Any]:
    """Computes the logic within and returns the value.

    Args:
        args: any arguments passed.
        kwargs: any key-word arguments passed

    Returns:
        the computed output of this expression or another expression with a partial operation \
        in it if the args provided are less than expected.
    """
    value = self._f(*args, **kwargs)
    if isinstance(value, Operation):
        return Expression(f=value)
    return value

__rshift__(nxt)

This makes piping using the '>>' symbol possible.

Combines with the given nxt expression or pipeline to produce a new pipeline where data flows from current to nxt.

Parameters:

Name	Type	Description	Default
nxt	Union[Expression, Pipeline, Callable]	the next expression, pipeline, or callable to apply after the current one.	required

Returns:

Type	Description
Union[Pipeline, Any]	a new pipeline where the first expression is the current expression followed by nxt or returns the value when the pipeline is executed in case nxt is of type ExecutionExpression

Source code in funml/types.py

def __rshift__(
    self,
    nxt: Union[
        "Expression",
        "Pipeline",
        Callable,
    ],
) -> Union["Pipeline", Any]:
    """This makes piping using the '>>' symbol possible.

    Combines with the given `nxt` expression or pipeline to produce a new pipeline
    where data flows from current to nxt.

    Args:
        nxt: the next expression, pipeline, or callable to apply after the current one.

    Returns:
        a new pipeline  where the first expression is the current expression followed by `nxt` \
        or returns the value when the pipeline is executed in case `nxt` is of type `ExecutionExpression`
    """
    new_pipeline = Pipeline()
    new_pipeline >> self >> nxt
    return new_pipeline

ExecutionExpression

Bases: Expression

Expression that executes all previous once it is found on a pipeline

Raises:

Type	Description
NotImplementedError	when >> is used after it.

Source code in funml/types.py

class ExecutionExpression(Expression):
    """Expression that executes all previous once it is found on a pipeline

    Raises:
        NotImplementedError: when `>>` is used after it.
    """

    def __rshift__(self, nxt: Any):
        """rshift is not supported for this.

        This is a terminal expression that expects no other expression
        after it on the pipeline.
        """
        raise NotImplementedError("terminal pipeline expression: `>>` not supported")

__rshift__(nxt)

rshift is not supported for this.

This is a terminal expression that expects no other expression after it on the pipeline.

Source code in funml/types.py

def __rshift__(self, nxt: Any):
    """rshift is not supported for this.

    This is a terminal expression that expects no other expression
    after it on the pipeline.
    """
    raise NotImplementedError("terminal pipeline expression: `>>` not supported")

MatchExpression

Bases: Expression

A special expression used when pattern matching.

Parameters:

Name	Type	Description	Default
arg	Optional[Any]	the value to be pattern matched.	None

Source code in funml/types.py

class MatchExpression(Expression):
    """A special expression used when pattern matching.

    Args:
        arg: the value to be pattern matched.
    """

    def __init__(self, arg: Optional[Any] = None):
        super().__init__()
        self._matches: List[Tuple[Callable[[Any], bool], Expression[T]]] = []
        self.__arg = arg

    def case(self, pattern: Union[MLType, Any], do: Callable) -> "MatchExpression":
        """Adds a case to a match statement.

        This is chainable, allowing multiple cases to be added to the same
        match pipeline.

        Args:
            pattern: the pattern to match against.
            do: the logic to run if pattern is matched.

        Returns:
            The current match expressions, after adding the case.
        """
        if isinstance(pattern, MLType):
            check, expn = pattern.generate_case(Operation(func=do))
        else:
            check = lambda arg: is_equal_or_of_type(arg, pattern)
            expn = Expression(Operation(func=do))

        self.__add_match(check=check, expn=expn)
        return self

    def __add_match(self, check: Callable[[Any], bool], expn: "Expression"):
        """Adds a match set to the list of match sets

        A match set comprises a checker function and an expression.
        The checker function checks if a given argument matches this case.
        The expression is called when the case is matched.

        Args:
            check: the checker function
            expn: the expression to run if a value matches.
        """
        if not callable(check):
            raise TypeError(f"the check is supposed to be a callable. Got {check}")

        if not isinstance(expn, Expression):
            raise TypeError(
                f"expected expression to be an Expression. Got {type(expn)}"
            )

        self._matches.append((check, expn))

    def __call__(self, arg: Optional[Any] = None) -> Union["Expression", Any]:
        """Applies the matched case and returns the output.

        The match cases are surveyed for any that matches the given argument
        until one that matches is found.
        Then the expression of that case is run and its output returned.

        Args:
            arg: the potential value to match against.

        Returns:
            The output of the expression of the matched case.

        Raises:
            MatchError: no case was matched for the given argument.
        """
        if arg is None:
            arg = self.__arg

        for check, expn in self._matches:
            if check(arg):
                return expn(arg)

        raise errors.MatchError(arg)

__add_match(check, expn)

Adds a match set to the list of match sets

A match set comprises a checker function and an expression. The checker function checks if a given argument matches this case. The expression is called when the case is matched.

Parameters:

Name	Type	Description	Default
check	Callable[[Any], bool]	the checker function	required
expn	Expression	the expression to run if a value matches.	required

Source code in funml/types.py

def __add_match(self, check: Callable[[Any], bool], expn: "Expression"):
    """Adds a match set to the list of match sets

    A match set comprises a checker function and an expression.
    The checker function checks if a given argument matches this case.
    The expression is called when the case is matched.

    Args:
        check: the checker function
        expn: the expression to run if a value matches.
    """
    if not callable(check):
        raise TypeError(f"the check is supposed to be a callable. Got {check}")

    if not isinstance(expn, Expression):
        raise TypeError(
            f"expected expression to be an Expression. Got {type(expn)}"
        )

    self._matches.append((check, expn))

__call__(arg=None)

Applies the matched case and returns the output.

The match cases are surveyed for any that matches the given argument until one that matches is found. Then the expression of that case is run and its output returned.

Parameters:

Name	Type	Description	Default
arg	Optional[Any]	the potential value to match against.	None

Returns:

Type	Description
Union[Expression, Any]	The output of the expression of the matched case.

Raises:

Type	Description
MatchError	no case was matched for the given argument.

Source code in funml/types.py

def __call__(self, arg: Optional[Any] = None) -> Union["Expression", Any]:
    """Applies the matched case and returns the output.

    The match cases are surveyed for any that matches the given argument
    until one that matches is found.
    Then the expression of that case is run and its output returned.

    Args:
        arg: the potential value to match against.

    Returns:
        The output of the expression of the matched case.

    Raises:
        MatchError: no case was matched for the given argument.
    """
    if arg is None:
        arg = self.__arg

    for check, expn in self._matches:
        if check(arg):
            return expn(arg)

    raise errors.MatchError(arg)

case(pattern, do)

Adds a case to a match statement.

This is chainable, allowing multiple cases to be added to the same match pipeline.

Parameters:

Name	Type	Description	Default
pattern	Union[MLType, Any]	the pattern to match against.	required
do	Callable	the logic to run if pattern is matched.	required

Returns:

Type	Description
MatchExpression	The current match expressions, after adding the case.

Source code in funml/types.py

def case(self, pattern: Union[MLType, Any], do: Callable) -> "MatchExpression":
    """Adds a case to a match statement.

    This is chainable, allowing multiple cases to be added to the same
    match pipeline.

    Args:
        pattern: the pattern to match against.
        do: the logic to run if pattern is matched.

    Returns:
        The current match expressions, after adding the case.
    """
    if isinstance(pattern, MLType):
        check, expn = pattern.generate_case(Operation(func=do))
    else:
        check = lambda arg: is_equal_or_of_type(arg, pattern)
        expn = Expression(Operation(func=do))

    self.__add_match(check=check, expn=expn)
    return self

MLType

An ML-enabled type, that can easily be used in pattern matching, piping etc.

Methods common to ML-enabled types are defined in this class.

Source code in funml/types.py

class MLType:
    """An ML-enabled type, that can easily be used in pattern matching, piping etc.

    Methods common to ML-enabled types are defined in this class.
    """

    def generate_case(
        self, do: "Operation"
    ) -> Tuple[Callable[[Any], bool], "Expression"]:
        """Generates a case statement for pattern matching.

        Args:
            do: The operation to do if the arg matches on this type

        Returns:
            A tuple (checker, expn) where checker is a function that checks if argument matches this case, and expn \
            is the expression that is called when the case is matched.
        """
        raise NotImplemented("generate_case not implemented")

    def _is_like(self, other: Any) -> bool:
        """Checks whether a value has the given pattern.

        Args:
            other: the value being checked against the pattern represented by type.

        Returns:
            A boolean showing true if `other` matches the pattern represented by current type.
        """
        raise NotImplemented("_is_like not implemented")

generate_case(do)

Generates a case statement for pattern matching.

Parameters:

Name	Type	Description	Default
do	Operation	The operation to do if the arg matches on this type	required

Returns:

Type	Description
Tuple[Callable[[Any], bool], Expression]	A tuple (checker, expn) where checker is a function that checks if argument matches this case, and expn is the expression that is called when the case is matched.

Source code in funml/types.py

def generate_case(
    self, do: "Operation"
) -> Tuple[Callable[[Any], bool], "Expression"]:
    """Generates a case statement for pattern matching.

    Args:
        do: The operation to do if the arg matches on this type

    Returns:
        A tuple (checker, expn) where checker is a function that checks if argument matches this case, and expn \
        is the expression that is called when the case is matched.
    """
    raise NotImplemented("generate_case not implemented")

Operation

A computation.

Parameters:

Name	Type	Description	Default
func	Callable	the logic to run as part of the operation.	required

Source code in funml/types.py

class Operation:
    """A computation.

    Args:
        func: the logic to run as part of the operation.
    """

    def __init__(self, func: Callable):
        self.__signature = _get_func_signature(func)
        self.__args_length = _get_non_variable_args_length(self.__signature)

        if len(self.__signature.parameters) == 0:
            # be more fault tolerant by using variable params
            self.__f = lambda *args, **kwargs: func()
        else:
            self.__f = func

    def __call__(self, *args: Any, **kwargs: Any) -> Union["Operation", Any]:
        """Applies the logic attached to this operation and returns output.

        Args:
            args: the args passed
            kwargs: the context in which the operation is being run.

        Returns:
            the final output of the operation's logic code or a partial operation if the args and kwargs \
            provided are less than those expected.
        """
        try:
            args_length = _get_num_of_relevant_args(self.__signature, *args, **kwargs)
            if args_length < self.__args_length:
                return Operation(func=functools.partial(self.__f, *args, **kwargs))
        except TypeError:
            # binding is impossible so just use the default implementation
            pass

        return self.__f(*args, **kwargs)

__call__(*args, **kwargs)

Applies the logic attached to this operation and returns output.

Parameters:

Name	Type	Description	Default
args	Any	the args passed	()
kwargs	Any	the context in which the operation is being run.	{}

Returns:

Type	Description
Union[Operation, Any]	the final output of the operation's logic code or a partial operation if the args and kwargs provided are less than those expected.

Source code in funml/types.py

def __call__(self, *args: Any, **kwargs: Any) -> Union["Operation", Any]:
    """Applies the logic attached to this operation and returns output.

    Args:
        args: the args passed
        kwargs: the context in which the operation is being run.

    Returns:
        the final output of the operation's logic code or a partial operation if the args and kwargs \
        provided are less than those expected.
    """
    try:
        args_length = _get_num_of_relevant_args(self.__signature, *args, **kwargs)
        if args_length < self.__args_length:
            return Operation(func=functools.partial(self.__f, *args, **kwargs))
    except TypeError:
        # binding is impossible so just use the default implementation
        pass

    return self.__f(*args, **kwargs)

Errors

Errors for this domain

MatchError

Bases: BaseException

Exception returned when a match fails to find an appropriate case for argument.

Parameters:

Name	Type	Description	Default
arg	str	the argument whose match was not found	required

Source code in funml/errors.py

class MatchError(BaseException):
    """Exception returned when a match fails to find an appropriate case for argument.

    Args:
        arg: the argument whose match was not found
    """

    def __init__(self, arg: str):
        self.arg = arg

    def __repr__(self):
        return f"MatchError: No match found for {self.arg}"