Skip to content

FuturePool

FuturePool 

FuturePool(number_of_workers: int = cpu_count() or 1)

Class representing a pool of async workers to simplify working with async functions that need to be restrained. Common use is fetching data from sites, where in order to not influence the website performance, limit on active connections is set.

Source code in futurepool/futurepool.py 
36
37
38
39
40
41
42
def __init__(self, number_of_workers: int = (os.cpu_count() or 1)):
    """ """
    assert number_of_workers > 0, "Number of workers must be a positive number"
    self.number_of_workers = number_of_workers
    self.loop = get_running_loop()
    self.workers_locks = [asyncio.Lock() for _ in range(self.number_of_workers)]
    self.tasks = set[Task]()

imap 

imap(fn: Callable[[T], Awaitable[U]], iterable: Iterable[T]) -> Iterator[Awaitable[U]]

Lazy equivalent of map(). Returns iterator of futures.

Examples: 

async with FuturePool(3) as fp:
    iterator = fp.imap(async_fn, range(10000))
    a = await next(iterator)
    b = await next(iterator)
    c = await next(iterator)

async with FuturePool(3) as fp:
    for future in fp.imap(async_fn, range(10000)):
        result = await future
        do_sth(result)

Source code in futurepool/futurepool.py 
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
def imap(
    self, fn: Callable[[T], Awaitable[U]], iterable: Iterable[T]
) -> Iterator[Awaitable[U]]:
    """
    Lazy equivalent of `map()`. Returns iterator of futures.

    Examples:
    ``` python
    async with FuturePool(3) as fp:
        iterator = fp.imap(async_fn, range(10000))
        a = await next(iterator)
        b = await next(iterator)
        c = await next(iterator)

    async with FuturePool(3) as fp:
        for future in fp.imap(async_fn, range(10000)):
            result = await future
            do_sth(result)
    ```
    """

    return self._get_iterator_(fn, ((arg,) for arg in iterable), True)

imap_async 

imap_async(fn: Callable[[T], Awaitable[U]], iterable: Iterable[T]) -> AsyncIterator[U]

Lazy async equivalent of map(). Returns async iterator of U.

Example: 

async with FuturePool(3) as fp:
    async for result in fp.imap_async(async_fn, range(10000)):
        do_sth(result)

Source code in futurepool/futurepool.py 
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
def imap_async(
    self, fn: Callable[[T], Awaitable[U]], iterable: Iterable[T]
) -> AsyncIterator[U]:
    """
    Lazy async equivalent of `map()`. Returns async iterator of U.

    Example:
    ``` python
    async with FuturePool(3) as fp:
        async for result in fp.imap_async(async_fn, range(10000)):
            do_sth(result)
    ```
    """

    return FuturePool.IteratorToAsyncIterator(self.imap(fn, iterable))

imap_unordered 

imap_unordered(fn: Callable[[T], Awaitable[U]], iterable: Iterable[T]) -> Iterator[Awaitable[U]]

The same as imap() except that the ordering of the results from the returned iterator should be considered arbitrary. (Only when there is only one worker process is the order guaranteed to be 'correct'.)

Examples: 

async with FuturePool(3) as fp:
    iterator = fp.imap_unordered(async_fn_that_takes_variable_time, range(10000))
    a = await next(iterator) // could be async_fn_that_takes_variable_time(0) or async_fn_that_takes_variable_time(1) or ...
    b = await next(iterator) // could be async_fn_that_takes_variable_time(0) or async_fn_that_takes_variable_time(1) or ...
    c = await next(iterator) // could be async_fn_that_takes_variable_time(0) or async_fn_that_takes_variable_time(1) or ...

async with FuturePool(3) as fp:
    for future in fp.imap_unordered(async_fn_that_takes_variable_time, range(10000)):
        result = await future // could be async_fn_that_takes_variable_time(0) or async_fn_that_takes_variable_time(1) or ...
        do_sth(result)

Source code in futurepool/futurepool.py 
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
def imap_unordered(
    self, fn: Callable[[T], Awaitable[U]], iterable: Iterable[T]
) -> Iterator[Awaitable[U]]:
    """
    The same as `imap()` except that the ordering of the results from the returned iterator should be considered arbitrary.
    (Only when there is only one worker process is the order guaranteed to be 'correct'.)

    Examples:
    ``` python
    async with FuturePool(3) as fp:
        iterator = fp.imap_unordered(async_fn_that_takes_variable_time, range(10000))
        a = await next(iterator) // could be async_fn_that_takes_variable_time(0) or async_fn_that_takes_variable_time(1) or ...
        b = await next(iterator) // could be async_fn_that_takes_variable_time(0) or async_fn_that_takes_variable_time(1) or ...
        c = await next(iterator) // could be async_fn_that_takes_variable_time(0) or async_fn_that_takes_variable_time(1) or ...

    async with FuturePool(3) as fp:
        for future in fp.imap_unordered(async_fn_that_takes_variable_time, range(10000)):
            result = await future // could be async_fn_that_takes_variable_time(0) or async_fn_that_takes_variable_time(1) or ...
            do_sth(result)
    ```
    """
    return self._get_iterator_(fn, ((arg,) for arg in iterable), False)

imap_unordered_async 

imap_unordered_async(fn: Callable[[T], Awaitable[U]], iterable: Iterable[T]) -> AsyncIterator[U]

The same as imap_async() except that the ordering of the results from the returned iterator should be considered arbitrary. (Only when there is only one worker process is the order guaranteed to be 'correct'.)

Example: 

async with FuturePool(3) as fp:
    async for result in fp.imap_unordered_async(async_fn_that_takes_variable_time, range(10000)):
        // result could be async_fn_that_takes_variable_time(0) or async_fn_that_takes_variable_time(1) or ...
        do_sth(result)

Source code in futurepool/futurepool.py 
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
def imap_unordered_async(
    self, fn: Callable[[T], Awaitable[U]], iterable: Iterable[T]
) -> AsyncIterator[U]:
    """
    The same as `imap_async()` except that the ordering of the results from the returned iterator should be considered arbitrary.
    (Only when there is only one worker process is the order guaranteed to be 'correct'.)

    Example:
    ``` python
    async with FuturePool(3) as fp:
        async for result in fp.imap_unordered_async(async_fn_that_takes_variable_time, range(10000)):
            // result could be async_fn_that_takes_variable_time(0) or async_fn_that_takes_variable_time(1) or ...
            do_sth(result)
    ```
    """
    return FuturePool.IteratorToAsyncIterator(self.imap_unordered(fn, iterable))

map 

map(fn: Callable[[T], Awaitable[U]], iterable: Iterable[T]) -> Awaitable[list[U]]

Parallel equivalent of standard map function. Applies provided fn on each item in iterable utilizing number_of_workers workers. Function returns Future of all results.

Example: 

async with FuturePool(3) as fp:
    result = await fp.map(async_fn, range(3))

Source code in futurepool/futurepool.py 
134
135
136
137
138
139
140
141
142
143
144
145
146
147
def map(
    self, fn: Callable[[T], Awaitable[U]], iterable: Iterable[T]
) -> Awaitable[list[U]]:
    """Parallel equivalent of standard map function. Applies provided fn on each item in iterable
    utilizing number_of_workers workers. Function returns Future of all results.

    Example:
    ``` python
    async with FuturePool(3) as fp:
        result = await fp.map(async_fn, range(3))
    ```

    """
    return self.starmap(fn, ((arg,) for arg in iterable))

starimap 

starimap(fn: Callable[[Any], Awaitable[U]], iterable: Iterable[Iterable[Any]]) -> Iterator[Awaitable[U]]

Like imap() except that the elements of the iterable are expected to be iterables that are unpacked as arguments.

Example: 

async with FuturePool(3) as fp:
    iterator = fp.starimap(async_fn_with_2_args, zip(range(10000), range(10000)))
    a = await next(iterator)
    b = await next(iterator)
    c = await next(iterator)

async with FuturePool(3) as fp:
    for future in fp.starimap(async_fn_with_2_args, zip(range(10000), range(10000))):
        result = await future
        do_sth(result)

Source code in futurepool/futurepool.py 
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
def starimap(
    self, fn: Callable[[Any], Awaitable[U]], iterable: Iterable[Iterable[Any]]
) -> Iterator[Awaitable[U]]:
    """
    Like `imap()` except that the elements of the iterable are expected to be iterables that are unpacked as arguments.

    Example:
    ``` python
    async with FuturePool(3) as fp:
        iterator = fp.starimap(async_fn_with_2_args, zip(range(10000), range(10000)))
        a = await next(iterator)
        b = await next(iterator)
        c = await next(iterator)

    async with FuturePool(3) as fp:
        for future in fp.starimap(async_fn_with_2_args, zip(range(10000), range(10000))):
            result = await future
            do_sth(result)
    ```
    """

    return self._get_iterator_(fn, iterable, True)

starimap_async 

starimap_async(fn: Callable[[Any], Awaitable[U]], iterable: Iterable[Iterable[Any]]) -> AsyncIterator[U]

Lazy async equivalent of starimap(). Returns async iterator of U.

Example: 

async with FuturePool(3) as fp:
    async for result in fp.starimap_async(async_fn_with_2_args, zip(range(10000), range(10000))):
        do_sth(result)

Source code in futurepool/futurepool.py 
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
def starimap_async(
    self, fn: Callable[[Any], Awaitable[U]], iterable: Iterable[Iterable[Any]]
) -> AsyncIterator[U]:
    """
    Lazy async equivalent of `starimap()`. Returns async iterator of U.

    Example:
    ``` python
    async with FuturePool(3) as fp:
        async for result in fp.starimap_async(async_fn_with_2_args, zip(range(10000), range(10000))):
            do_sth(result)
    ```
    """

    return FuturePool.IteratorToAsyncIterator(self.starimap(fn, iterable))

starimap_unordered 

starimap_unordered(fn: Callable[[T], Awaitable[U]], iterable: Iterable[T]) -> Iterator[Awaitable[U]]

The same as starimap() except that the ordering of the results from the returned iterator should be considered arbitrary. (Only when there is only one worker process is the order guaranteed to be 'correct'.)

Examples: 

async with FuturePool(3) as fp:
    iterator = fp.starimap_unordered(async_fn_that_takes_variable_time_with_2_args, zip(range(10000), range(10000)))
    a = await next(iterator) // could be async_fn_that_takes_variable_time_with_2_args(0, 0) or async_fn_that_takes_variable_time_with_2_args(1, 1) or ...
    b = await next(iterator) // could be async_fn_that_takes_variable_time_with_2_args(0, 0) or async_fn_that_takes_variable_time_with_2_args(1, 1) or ...
    c = await next(iterator) // could be async_fn_that_takes_variable_time_with_2_args(0, 0) or async_fn_that_takes_variable_time_with_2_args(1, 1) or ...

async with FuturePool(3) as fp:
    for future in fp.starimap_unordered(async_fn_that_takes_variable_time_with_2_args, zip(range(10000), range(10000))):
        result = await future // could be async_fn_that_takes_variable_time_with_2_args(0, 0) or async_fn_that_takes_variable_time_with_2_args(1, 1) or ...
        do_sth(result)

Source code in futurepool/futurepool.py 
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
def starimap_unordered(
    self, fn: Callable[[T], Awaitable[U]], iterable: Iterable[T]
) -> Iterator[Awaitable[U]]:
    """
    The same as `starimap()` except that the ordering of the results from the returned iterator should be considered arbitrary.
    (Only when there is only one worker process is the order guaranteed to be 'correct'.)

    Examples:
    ``` python
    async with FuturePool(3) as fp:
        iterator = fp.starimap_unordered(async_fn_that_takes_variable_time_with_2_args, zip(range(10000), range(10000)))
        a = await next(iterator) // could be async_fn_that_takes_variable_time_with_2_args(0, 0) or async_fn_that_takes_variable_time_with_2_args(1, 1) or ...
        b = await next(iterator) // could be async_fn_that_takes_variable_time_with_2_args(0, 0) or async_fn_that_takes_variable_time_with_2_args(1, 1) or ...
        c = await next(iterator) // could be async_fn_that_takes_variable_time_with_2_args(0, 0) or async_fn_that_takes_variable_time_with_2_args(1, 1) or ...

    async with FuturePool(3) as fp:
        for future in fp.starimap_unordered(async_fn_that_takes_variable_time_with_2_args, zip(range(10000), range(10000))):
            result = await future // could be async_fn_that_takes_variable_time_with_2_args(0, 0) or async_fn_that_takes_variable_time_with_2_args(1, 1) or ...
            do_sth(result)
    ```
    """
    return self._get_iterator_(fn, iterable, False)

starimap_unordered_async 

starimap_unordered_async(fn: Callable[[T], Awaitable[U]], iterable: Iterable[T]) -> AsyncIterator[U]

The same as starimap_async() except that the ordering of the results from the returned iterator should be considered arbitrary. (Only when there is only one worker process is the order guaranteed to be 'correct'.)

Example: 

async with FuturePool(3) as fp:
    async for result in fp.starimap_unordered_async(async_fn_that_takes_variable_time_with_2_args, zip(range(10000), range(10000))):
        // result could be async_fn_that_takes_variable_time_with_2_args(0, 0) or async_fn_that_takes_variable_time_with_2_args(1, 1) or ...
        do_sth(result)

Source code in futurepool/futurepool.py 
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
def starimap_unordered_async(
    self, fn: Callable[[T], Awaitable[U]], iterable: Iterable[T]
) -> AsyncIterator[U]:
    """
    The same as `starimap_async()` except that the ordering of the results from the returned iterator should be considered arbitrary.
    (Only when there is only one worker process is the order guaranteed to be 'correct'.)

    Example:
    ``` python
    async with FuturePool(3) as fp:
        async for result in fp.starimap_unordered_async(async_fn_that_takes_variable_time_with_2_args, zip(range(10000), range(10000))):
            // result could be async_fn_that_takes_variable_time_with_2_args(0, 0) or async_fn_that_takes_variable_time_with_2_args(1, 1) or ...
            do_sth(result)
    ```
    """
    return FuturePool.IteratorToAsyncIterator(self.starimap_unordered(fn, iterable))

starmap 

starmap(fn: Callable[[Any], Awaitable[U]], iterable: Iterable[Iterable[Any]]) -> Awaitable[list[U]]

Like map() except that the elements of the iterable are expected to be iterables that are unpacked as arguments. Hence an iterable of [(1,2), (3, 4)] results in [func(1,2), func(3,4)].

Example: 

async with FuturePool(3) as fp:
    result = await fp.starmap(async_fn_with_2_args, zip(range(3), range(3)))

Source code in futurepool/futurepool.py 
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
def starmap(
    self, fn: Callable[[Any], Awaitable[U]], iterable: Iterable[Iterable[Any]]
) -> Awaitable[list[U]]:
    """
    Like `map()` except that the elements of the iterable are expected to be iterables that are unpacked as arguments.
    Hence an iterable of `[(1,2), (3, 4)]` results in `[func(1,2), func(3,4)]`.

    Example:
    ``` python
    async with FuturePool(3) as fp:
        result = await fp.starmap(async_fn_with_2_args, zip(range(3), range(3)))
    ```
    """

    async def lazy_resolve():
        return [await future for future in self._get_iterator_(fn, iterable, True)]

    return lazy_resolve()