Skip to content

Future

Asynchronous computation using futures.

future_value = Future.from_value module-attribute

An alias of Future.from_value.

Future

Bases: Awaitable[T]

Represents future computations.

Source code in src/wraps/futures/future.py 
 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
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
@frozen()
class Future(Awaitable[T]):
    """Represents future computations."""

    awaitable: ReAwaitable[T] = field(converter=ReAwaitable)

    def __repr__(self) -> str:
        return empty_repr(self)

    def __await__(self) -> Generator[None, None, T]:
        return self.awaitable.__await__()

    @classmethod
    def create(cls, awaitable: Awaitable[U]) -> Future[U]:
        """Creates a [`Future[U]`][wraps.futures.future.Future]
        from an [`Awaitable[U]`][typing.Awaitable].

        Arguments:
            awaitable: The awaitable to wrap.

        Returns:
            The future wrapping the given awaitable.
        """
        return cls(awaitable)  # type: ignore[arg-type, return-value]

    def future_map(self, function: Unary[T, U]) -> Future[U]:
        """Maps a [`Future[T]`][wraps.futures.future.Future] to a [`Future[U]`][wraps.futures.future.Future]
        by applying the `function` to its result.

        Arguments:
            function: The function to apply.

        Returns:
            The mapped future.
        """
        return self.create(self.raw_future_map(function))

    def future_map_await(self, function: AsyncUnary[T, U]) -> Future[U]:
        """Maps a [`Future[T]`][wraps.futures.future.Future]
        to a [`Future[U]`][wraps.futures.future.Future]
        by applying the asynchronous `function` to its result.

        Arguments:
            function: The asynchronous function to apply.

        Returns:
            The mapped future.
        """
        return self.create(self.raw_future_map_await(function))

    async def raw_future_map(self, function: Unary[T, U]) -> U:
        return function(await self.awaitable)

    async def raw_future_map_await(self, function: AsyncUnary[T, U]) -> U:
        return await function(await self.awaitable)

    def then(self, function: FutureUnary[T, U]) -> Future[U]:
        """Chains computation by applying the `function` to the result, returning the resulting
        [`Future[U]`][wraps.futures.future.Future].

        Arguments:
            function: The future-returning function to apply.

        Returns:
            The resulting future.
        """
        return self.create(self.raw_then(function))

    async def raw_then(self, function: FutureUnary[T, U]) -> U:
        return await function(await self.awaitable).awaitable

    def future_flatten(self: Future[Future[U]]) -> Future[U]:
        """Flattens a [`Future[Future[U]]`][wraps.futures.future.Future]
        to a [`Future[U]`][wraps.futures.future.Future].

        This is identical to:

        ```python
        future.then(identity)
        ```

        Returns:
            The flattened future.
        """
        return self.then(identity)

    def __aiter__(self) -> AsyncIterator[T]:
        return self.async_iter()

    async def async_iter(self) -> AsyncIterator[T]:
        """Creates an asynchronous iterator yielding the result of this
        [`Future[T]`][wraps.futures.future.Future].

        Returns:
            An asynchronous iterator yielding the result of the future.
        """
        yield await self.awaitable

    @classmethod
    def from_value(cls, value: U) -> Future[U]:
        """Wraps the `value` of type `U` into a [`Future[U]`][wraps.futures.future.Future].

        This is functionally the same as:

        ```python
        async def async_identity(value: T) -> T:
            return value

        value = 42

        future = Future(async_identity(value))
        ```

        Example:
            ```python
            value = 42

            future = Future.from_value(value)

            assert await future is value
            ```

        Arguments:
            value: The value to wrap.

        Returns:
            The future wrapping the given value.
        """
        return cls.create(async_identity(value))

create(awaitable: Awaitable[U]) -> Future[U] classmethod

Creates a Future[U] from an Awaitable[U].

Parameters:

Name Type Description Default
awaitable Awaitable[U]

The awaitable to wrap.

 required

Returns:

Type Description
Future[U]

The future wrapping the given awaitable.
Source code in src/wraps/futures/future.py 
39
40
41
42
43
44
45
46
47
48
49
50
@classmethod
def create(cls, awaitable: Awaitable[U]) -> Future[U]:
    """Creates a [`Future[U]`][wraps.futures.future.Future]
    from an [`Awaitable[U]`][typing.Awaitable].

    Arguments:
        awaitable: The awaitable to wrap.

    Returns:
        The future wrapping the given awaitable.
    """
    return cls(awaitable)  # type: ignore[arg-type, return-value]

future_map(function: Unary[T, U]) -> Future[U]

Maps a Future[T] to a Future[U] by applying the function to its result.

Parameters:

Name Type Description Default
function Unary[T, U]

The function to apply.

 required

Returns:

Type Description
Future[U]

The mapped future.
Source code in src/wraps/futures/future.py 
52
53
54
55
56
57
58
59
60
61
62
def future_map(self, function: Unary[T, U]) -> Future[U]:
    """Maps a [`Future[T]`][wraps.futures.future.Future] to a [`Future[U]`][wraps.futures.future.Future]
    by applying the `function` to its result.

    Arguments:
        function: The function to apply.

    Returns:
        The mapped future.
    """
    return self.create(self.raw_future_map(function))

future_map_await(function: AsyncUnary[T, U]) -> Future[U]

Maps a Future[T] to a Future[U] by applying the asynchronous function to its result.

Parameters:

Name Type Description Default
function AsyncUnary[T, U]

The asynchronous function to apply.

 required

Returns:

Type Description
Future[U]

The mapped future.
Source code in src/wraps/futures/future.py 
64
65
66
67
68
69
70
71
72
73
74
75
def future_map_await(self, function: AsyncUnary[T, U]) -> Future[U]:
    """Maps a [`Future[T]`][wraps.futures.future.Future]
    to a [`Future[U]`][wraps.futures.future.Future]
    by applying the asynchronous `function` to its result.

    Arguments:
        function: The asynchronous function to apply.

    Returns:
        The mapped future.
    """
    return self.create(self.raw_future_map_await(function))

then(function: FutureUnary[T, U]) -> Future[U]

Chains computation by applying the function to the result, returning the resulting Future[U].

Parameters:

Name Type Description Default
function FutureUnary[T, U]

The future-returning function to apply.

 required

Returns:

Type Description
Future[U]

The resulting future.
Source code in src/wraps/futures/future.py 
83
84
85
86
87
88
89
90
91
92
93
def then(self, function: FutureUnary[T, U]) -> Future[U]:
    """Chains computation by applying the `function` to the result, returning the resulting
    [`Future[U]`][wraps.futures.future.Future].

    Arguments:
        function: The future-returning function to apply.

    Returns:
        The resulting future.
    """
    return self.create(self.raw_then(function))

future_flatten() -> Future[U]

Flattens a Future[Future[U]] to a Future[U].

This is identical to:

future.then(identity)

Returns:

Type Description
Future[U]

The flattened future.
Source code in src/wraps/futures/future.py 
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
def future_flatten(self: Future[Future[U]]) -> Future[U]:
    """Flattens a [`Future[Future[U]]`][wraps.futures.future.Future]
    to a [`Future[U]`][wraps.futures.future.Future].

    This is identical to:

    ```python
    future.then(identity)
    ```

    Returns:
        The flattened future.
    """
    return self.then(identity)

async_iter() -> AsyncIterator[T] async

Creates an asynchronous iterator yielding the result of this Future[T].

Returns:

Type Description
AsyncIterator[T]

An asynchronous iterator yielding the result of the future.
Source code in src/wraps/futures/future.py 
116
117
118
119
120
121
122
123
async def async_iter(self) -> AsyncIterator[T]:
    """Creates an asynchronous iterator yielding the result of this
    [`Future[T]`][wraps.futures.future.Future].

    Returns:
        An asynchronous iterator yielding the result of the future.
    """
    yield await self.awaitable

from_value(value: U) -> Future[U] classmethod

Wraps the value of type U into a Future[U].

This is functionally the same as:

async def async_identity(value: T) -> T:
    return value

value = 42

future = Future(async_identity(value))
Example 
value = 42

future = Future.from_value(value)

assert await future is value

Parameters:

Name Type Description Default
value U

The value to wrap.

 required

Returns:

Type Description
Future[U]

The future wrapping the given value.
Source code in src/wraps/futures/future.py 
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
@classmethod
def from_value(cls, value: U) -> Future[U]:
    """Wraps the `value` of type `U` into a [`Future[U]`][wraps.futures.future.Future].

    This is functionally the same as:

    ```python
    async def async_identity(value: T) -> T:
        return value

    value = 42

    future = Future(async_identity(value))
    ```

    Example:
        ```python
        value = 42

        future = Future.from_value(value)

        assert await future is value
        ```

    Arguments:
        value: The value to wrap.

    Returns:
        The future wrapping the given value.
    """
    return cls.create(async_identity(value))

wrap_future(function: AsyncCallable[P, T]) -> FutureCallable[P, T]

Wraps the asynchronous function returning T into the function returning Future[T].

Example 
@wrap_future
async def function() -> int:
    return 42

string = "42"

result = await function().future_map(str)

assert result == string

Parameters:

Name Type Description Default
function AsyncCallable[P, T]

The asynchronous function to wrap.

 required

Returns:

Type Description
FutureCallable[P, T]

The wrapping function.
Source code in src/wraps/futures/future.py 
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
def wrap_future(function: AsyncCallable[P, T]) -> FutureCallable[P, T]:
    """Wraps the asynchronous `function` returning `T` into the function
    returning [`Future[T]`][wraps.futures.future.Future].

    Example:
        ```python
        @wrap_future
        async def function() -> int:
            return 42

        string = "42"

        result = await function().future_map(str)

        assert result == string
        ```

    Arguments:
        function: The asynchronous function to wrap.

    Returns:
        The wrapping function.
    """

    @wraps(function)
    def wrap(*args: P.args, **kwargs: P.kwargs) -> Future[T]:
        return Future(function(*args, **kwargs))

    return wrap