Bases: Generic[A]
AIO provides abstraction for asynchronous IO.
Source code in funclift/types/aio.py
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64 | @dataclass
class AIO(Generic[A]):
"""AIO provides abstraction for asynchronous IO."""
unsafe_run: Callable[[], A]
@staticmethod
def raise_error(or_else: Callable[[], Exception]) -> AIO[Exception]:
return AIO.pure(or_else())
@staticmethod
def from_option(option: Option[A],
or_else: Callable[[], Exception]) -> AIO[A] | AIO[Exception]:
"""Creates an AIO from an Option.
Args:
option (Option[A]): the option from which to create an AIO.
or_else (Callable[[], Exception]): this function will be invoked
when the passed in Option is Nothing.
Returns:
AIO[A] | AIO[Exception]: _description_
"""
match option:
case Some(v):
return AIO.pure(v)
case Nothing():
return AIO.raise_error(or_else)
case _:
return AIO.raise_error(or_else)
@staticmethod
def pure(a: B) -> AIO[B]:
return AIO(lambda: a)
def fmap(self, f: Callable[[A], B]) -> AIO[B]:
return AIO(lambda: f(self.unsafe_run()))
def flatmap(self, f: Callable[[A], AIO[B]]) -> AIO[B]:
# Should not do this because it executes self.unsafe_run() right away.
# return f(self.unsafe_run())
def foo():
a: A = self.unsafe_run()
fa: AIO[B] = f(a)
return fa.unsafe_run()
return AIO(foo)
def ap(self, a: AIO[D]) -> AIO[E]:
return a.fmap(lambda x: cast(Callable[[D], E], self.unsafe_run())(x))
|
from_option(option, or_else)
staticmethod
Creates an AIO from an Option.
Parameters:
| Name |
Type |
Description |
Default |
option |
Option[A]
|
the option from which to create an AIO. |
required
|
or_else |
Callable[[], Exception]
|
this function will be invoked
when the passed in Option is Nothing. |
required
|
Returns:
| Type |
Description |
AIO[A] | AIO[Exception]
|
AIO[A] | AIO[Exception]: description |
Source code in funclift/types/aio.py
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45 | @staticmethod
def from_option(option: Option[A],
or_else: Callable[[], Exception]) -> AIO[A] | AIO[Exception]:
"""Creates an AIO from an Option.
Args:
option (Option[A]): the option from which to create an AIO.
or_else (Callable[[], Exception]): this function will be invoked
when the passed in Option is Nothing.
Returns:
AIO[A] | AIO[Exception]: _description_
"""
match option:
case Some(v):
return AIO.pure(v)
case Nothing():
return AIO.raise_error(or_else)
case _:
return AIO.raise_error(or_else)
|