Core
Auto-generated documentation for fundom.core module.
compose
dataclass(slots=True, init=False)
class compose(Generic[P, V]):
def __init__() -> None:
Function composition abstraction.
If no function is passed to composition than Exception would be raised on call.
Example
f: Callable[[int], int] = (
compose()
<< (lambda x: x + 1)
<< (lambda x: x ** 2)
)
See also
future
dataclass(slots=True, frozen=True)
class future(Generic[T]):
Abstraction over awaitable value to run in pipeline.
Example
result = await (
future(get_user_async)
<< if_role_is("moderator")
<< set_role("admin")
>> update_user_async
)
See also
future.returns
@staticmethod
def returns(func: Callable[P, Coroutine[Any, Any, T]]):
Wraps returned value of async function to future.
Return value of function is wrapped into future.
Example
@future.returns
async def some_async_func(x: int, y: int) -> str:
...
See also
pipe
dataclass(slots=True, frozen=True)
class pipe(Generic[T]):
Abstraction over some value to run in pipeline.
Example
result: int = (
pipe(12)
<< (lambda x: x + 1)
<< (lambda x: x**2)
<< (lambda x: x // 3)
).finish()
See also
pipe().finish
def finish() -> T:
Finish pipe by unpacking internal value.
Examples
result = (
pipe(3)
<< (lambda x: x + 1)
<< (lambda x: x**2)
)
value = result.finish()
Returns
T- internal value
See also
pipe.returns
@staticmethod
def returns(func: Callable[P, pipe[T]]) -> Callable[P, T]:
Decorator for functions that return pipe object for seamless unwrapping.
Example
@pipe.returns
def some_function(x: int) -> pipe[bool]:
return (
pipe(x)
<< (lambda x: x + 1)
<< some_when(lambda x: x > 10)
<< if_some_returns(True)
<< if_none_returns(False)
# returned type is bool
See also
cfilter
@hof1
def cfilter(
predicate: Callable[[A1], bool],
lst: Iterable[A1],
) -> Iterable[A1]:
Curried filter function.
Arguments
predicate (Callable[[A1], bool]): to filter with.
lstIterable[A1] - to filter.
Returns
Iterable[A1]- filtered iterable.
See also
cmap
@hof1
def cmap(mapper: Callable[[A1], A2], lst: Iterable[A1]) -> Iterable[A2]:
Curried map function.
Arguments
mapper (Callable[[A1], A2]): mapper for element of iterable.
lstIterable[A1] - to map.
Returns
Iterable[A2]- map result.
See also
foldl
@hof2
def foldl(
folder: Callable[[A1, A2], A1],
initial: A1,
lst: Iterable[A2],
) -> A1:
Curried reduce left function.
Arguments
folder (Callable[[A1, A2], A1]): aggregator.
initialA1 - initial aggregation value.lstIterable[A2] - data to reduce.
Returns
A1- reduction result.
See also
foldr
@hof2
def foldr(
folder: Callable[[A1, A2], A2],
initial: A2,
lst: Iterable[A1],
) -> A2:
Curried reduce right function.
Arguments
folder (Callable[[A1, A2], A2]): aggregator.
initialA2 - initial aggregation value.lstIterable[A1] - data to reduce.
Returns
A2- reduction result.
See also
hof1
def hof1(f: Callable[Concatenate[A1, P], AResult]):
Separate first argument from other.
See also
hof2
def hof2(f: Callable[Concatenate[A1, A2, P], AResult]):
Separate first 2 arguments from other.
See also
hof3
def hof3(f: Callable[Concatenate[A1, A2, A3, P], AResult]):
Separate first 3 arguments from other.
See also
returns
def returns(x: T) -> Callable[P, T]:
Return T on any input.
Example
get_none: Callable[..., None] = returns(None)
See also
returns_future
def returns_future(x: T) -> Callable[P, future[T]]:
Return awaitable T on any input.
Example
get_none_future: Callable[..., future[None]] = returns_future(None)
See also
this
def this(args: T) -> T:
Synchronous identity function.
Example
this(3) # 3
See also
this_future
@future.returns
async def this_future(args: T) -> T:
Asynchronous identity function.
Example
this_future(3) # future(3)