Module raffiot.val

Local Variables to work around Python annoying limitations about lambdas.

Python forbids local variables and

"""
Local Variables to work around Python annoying limitations about lambdas.

Python forbids local variables and
"""
from __future__ import annotations

from collections import abc
from dataclasses import dataclass
from typing import Generic, TypeVar, Callable, List, Any

from typing_extensions import final

from raffiot import io, resource
from raffiot.io import IO
from raffiot.resource import Resource

__all__ = [
    "Val",
]


A = TypeVar("A")
B = TypeVar("B")


@final
@dataclass
class Val(Generic[A]):
    """
    Immutable Value.

    Used to create local "variables" in lambdas.
    """

    __slots__ = "value"

    value: A

    def get(self) -> A:
        """
        Get this Val value.
        :return:
        """
        return self.value

    def get_io(self) -> IO[None, None, A]:
        """
        Get this Val value.
        :return:
        """
        return io.defer(self.get)

    def get_rs(self) -> Resource[None, None, A]:
        """
        Get this Val value.
        :return:
        """
        return resource.defer(self.get)

    @classmethod
    def pure(cls, a: A) -> Val[A]:
        """
        Create a new Val with value `a`

        :param a: the value of this val.
        :return:
        """
        return Val(a)

    def map(self, f: Callable[[A], B]) -> Val[B]:
        """
        Create a new Val from this one by applying this **pure** function.

        :param f:
        :return:
        """
        return Val(f(self.value))

    def traverse(self, f: Callable[[A], IO[Any, Any, B]]) -> IO[Any, Any, Val[B]]:
        """
        Create a new Val from this one by applying this `IO` function.

        :param f:
        :return:
        """

        return io.defer_io(f, self.value).map(Val)

    def flat_map(self, f: Callable[[A], Val[B]]) -> Val[B]:
        """
        Create a new Val from this one.

        :param f:
        :return:
        """
        return f(self.value)

    def flatten(self: Val[Val[B]]) -> Val[B]:  # A = Val[B]
        """ "
        Flatten this `Val[Val[A]]` into a `Val[A]`
        """

        return Val(self.value.value)

    @classmethod
    def zip(cls, *vals: Val[A]) -> Val[List[A]]:
        """ "
        Group these list of Val into a Val of List
        """

        if len(vals) == 1 and isinstance(vals[0], abc.Iterable):
            return Val([x.value for x in vals[0]])
        return Val([x.value for x in vals])  # type: ignore

    def zip_with(self, *vals: Any) -> Val[List[A]]:
        """
        Group this Val with other Val into a list of Val.

        :param vals: other Val to combine with self.
        :return:
        """

        return Val.zip(self, *vals)

    def ap(self, *arg: Val[A]) -> Val[B]:
        """
        Apply the function contained in this Val to `args` Vals.

        :param arg:
        :return:
        """

        if len(arg) == 1 and isinstance(arg[0], abc.Iterable):
            l = [x.value for x in arg[0]]
        else:
            l = [x.value for x in arg]
        return Val(self.value(*l))  # type: ignore

Classes

class Val (value: A)

Immutable Value.

Used to create local "variables" in lambdas.

Expand source code

@final
@dataclass
class Val(Generic[A]):
    """
    Immutable Value.

    Used to create local "variables" in lambdas.
    """

    __slots__ = "value"

    value: A

    def get(self) -> A:
        """
        Get this Val value.
        :return:
        """
        return self.value

    def get_io(self) -> IO[None, None, A]:
        """
        Get this Val value.
        :return:
        """
        return io.defer(self.get)

    def get_rs(self) -> Resource[None, None, A]:
        """
        Get this Val value.
        :return:
        """
        return resource.defer(self.get)

    @classmethod
    def pure(cls, a: A) -> Val[A]:
        """
        Create a new Val with value `a`

        :param a: the value of this val.
        :return:
        """
        return Val(a)

    def map(self, f: Callable[[A], B]) -> Val[B]:
        """
        Create a new Val from this one by applying this **pure** function.

        :param f:
        :return:
        """
        return Val(f(self.value))

    def traverse(self, f: Callable[[A], IO[Any, Any, B]]) -> IO[Any, Any, Val[B]]:
        """
        Create a new Val from this one by applying this `IO` function.

        :param f:
        :return:
        """

        return io.defer_io(f, self.value).map(Val)

    def flat_map(self, f: Callable[[A], Val[B]]) -> Val[B]:
        """
        Create a new Val from this one.

        :param f:
        :return:
        """
        return f(self.value)

    def flatten(self: Val[Val[B]]) -> Val[B]:  # A = Val[B]
        """ "
        Flatten this `Val[Val[A]]` into a `Val[A]`
        """

        return Val(self.value.value)

    @classmethod
    def zip(cls, *vals: Val[A]) -> Val[List[A]]:
        """ "
        Group these list of Val into a Val of List
        """

        if len(vals) == 1 and isinstance(vals[0], abc.Iterable):
            return Val([x.value for x in vals[0]])
        return Val([x.value for x in vals])  # type: ignore

    def zip_with(self, *vals: Any) -> Val[List[A]]:
        """
        Group this Val with other Val into a list of Val.

        :param vals: other Val to combine with self.
        :return:
        """

        return Val.zip(self, *vals)

    def ap(self, *arg: Val[A]) -> Val[B]:
        """
        Apply the function contained in this Val to `args` Vals.

        :param arg:
        :return:
        """

        if len(arg) == 1 and isinstance(arg[0], abc.Iterable):
            l = [x.value for x in arg[0]]
        else:
            l = [x.value for x in arg]
        return Val(self.value(*l))  # type: ignore

Ancestors

• typing.Generic

Static methods

def pure(a: A) ‑> Val[~A]

Create a new Val with value a

:param a: the value of this val. :return:

Expand source code

@classmethod
def pure(cls, a: A) -> Val[A]:
    """
    Create a new Val with value `a`

    :param a: the value of this val.
    :return:
    """
    return Val(a)

def zip(*vals: Val[A]) ‑> Val[typing.List[~A]]

" Group these list of Val into a Val of List

Expand source code

@classmethod
def zip(cls, *vals: Val[A]) -> Val[List[A]]:
    """ "
    Group these list of Val into a Val of List
    """

    if len(vals) == 1 and isinstance(vals[0], abc.Iterable):
        return Val([x.value for x in vals[0]])
    return Val([x.value for x in vals])  # type: ignore

Instance variables

var value : ~A

Return an attribute of instance, which is of type owner.

Methods

def ap(self, *arg: Val[A]) ‑> Val[~B]

Apply the function contained in this Val to args Vals.

:param arg: :return:

Expand source code

def ap(self, *arg: Val[A]) -> Val[B]:
    """
    Apply the function contained in this Val to `args` Vals.

    :param arg:
    :return:
    """

    if len(arg) == 1 and isinstance(arg[0], abc.Iterable):
        l = [x.value for x in arg[0]]
    else:
        l = [x.value for x in arg]
    return Val(self.value(*l))  # type: ignore

def flat_map(self, f: Callable[[A], Val[B]]) ‑> Val[~B]

Create a new Val from this one.

:param f: :return:

Expand source code

def flat_map(self, f: Callable[[A], Val[B]]) -> Val[B]:
    """
    Create a new Val from this one.

    :param f:
    :return:
    """
    return f(self.value)

def flatten(self: Val[Val[B]]) ‑> Val[~B]

" Flatten this Val[Val[A]] into a Val[A]

Expand source code

def flatten(self: Val[Val[B]]) -> Val[B]:  # A = Val[B]
    """ "
    Flatten this `Val[Val[A]]` into a `Val[A]`
    """

    return Val(self.value.value)

def get(self) ‑> ~A

Get this Val value. :return:

Expand source code

def get(self) -> A:
    """
    Get this Val value.
    :return:
    """
    return self.value

def get_io(self) ‑> IO[NoneType, NoneType, ~A]

Get this Val value. :return:

Expand source code

def get_io(self) -> IO[None, None, A]:
    """
    Get this Val value.
    :return:
    """
    return io.defer(self.get)

def get_rs(self) ‑> Resource[NoneType, NoneType, ~A]

Get this Val value. :return:

Expand source code

def get_rs(self) -> Resource[None, None, A]:
    """
    Get this Val value.
    :return:
    """
    return resource.defer(self.get)

def map(self, f: Callable[[A], B]) ‑> Val[~B]

Create a new Val from this one by applying this pure function.

:param f: :return:

Expand source code

def map(self, f: Callable[[A], B]) -> Val[B]:
    """
    Create a new Val from this one by applying this **pure** function.

    :param f:
    :return:
    """
    return Val(f(self.value))

def traverse(self, f: Callable[[A], IO[Any, Any, B]]) ‑> IO[typing.Any, typing.Any, Val[~B]]

Create a new Val from this one by applying this IO function.

:param f: :return:

Expand source code

def traverse(self, f: Callable[[A], IO[Any, Any, B]]) -> IO[Any, Any, Val[B]]:
    """
    Create a new Val from this one by applying this `IO` function.

    :param f:
    :return:
    """

    return io.defer_io(f, self.value).map(Val)

def zip_with(self, *vals: Any) ‑> Val[typing.List[~A]]

Group this Val with other Val into a list of Val.

:param vals: other Val to combine with self. :return:

Expand source code

def zip_with(self, *vals: Any) -> Val[List[A]]:
    """
    Group this Val with other Val into a list of Val.

    :param vals: other Val to combine with self.
    :return:
    """

    return Val.zip(self, *vals)