future-map
future-map is a Python library to use together with the official concurrent.futures module.
Why future-map?
Because it's difficult to deal with an infinite or huge input with concurrent.future.ThreadPoolExecutor and concurrent.future.ProcessPoolExecutor. See the following example.
from concurrent.futures import ThreadPoolExecutor
def make_input(length):
return range(length)
def make_infinite_input():
count = 0
while True:
yield count
count += 1
def process(value):
return value * 2
if __name__ == '__main__':
with ThreadPoolExecutor(max_workers=3) as executor:
# Works well
for value in executor.map(process, make_input(10)):
print('Doubled value:', value)
# This freezes the process and memory usage keeps growing
for value in executor.map(process, make_infinite_input()):
print('Doubled value:', value)
Installation
Use the package manager pip to install future-map.
$ pip install future-map
Usage
This library provides FutureMap. See the following example.
from future_map import FutureMap
from concurrent.futures import ThreadPoolExecutor
def make_infinite_input():
count = 0
while True:
yield count
count += 1
def process(value):
return value * 2
if __name__ == '__main__':
with ThreadPoolExecutor(max_workers=3) as executor:
fm = FutureMap(
lambda value: executor.submit(process, value),
make_infinite_input(), buffersize=5
)
for value in fm:
print('Doubled value:', value)
For more complicated use case:
import time
from concurrent.futures import ThreadPoolExecutor
from future_map import FutureMap
class APIClient:
def __init__(self, max_connections):
self.__max_connections = max_connections
self.__executor = None
def __enter__(self):
self.__executor = ThreadPoolExecutor(max_workers=self.__max_connections)
return self
def __exit__(self, exc_type, exc_value, traceback):
self.__executor.shutdown()
self.__executor = None
def call(self, url):
time.sleep(1)
return "Response from {}".format(url)
def call_async(self, url):
if self.__executor is None:
raise Exception("call_async needs to be called in the runtime context with this APIClient")
return self.__executor.submit(self.call, url)
def make_urls():
for i in range(100):
yield "https://example.com/api/resources/{}".format(i)
if __name__ == '__main__':
with APIClient(max_connections=3) as api_client:
for response in FutureMap(api_client.call_async, make_urls(), buffersize=5):
print(response)
API
FutureMap(fn, iterable, buffersize)
Constructor of FutureMap.
FutureMap is an iterable object that maps an iterable object (iterable argument) to a function (fn argument), waits until each future object is done, and yields each result.
Please note that this object will yield unordered results.
- Arguments
fn: Callable object that takes an argument from iterable, and return aconcurrent.futures.Future.iterable: Iterable object.buffersize: Maximum size of internal buffer. Eachconcurrent.futures.Futureobject is stored in the buffer until it's done. If the buffer is fulfilled,FutureMapstops reading values fromiterable.
- Return
FutureMapinstance
future_map(fn, iterable, buffersize)
Alias of FutureMap. You can use this function if you prefer a similar syntax with the map function.
For more details, please refer to FutureMap(fn, iterable, buffersize).
Contributing
Pull requests are welcome. For major changes, please open an issue first to discuss what you would like to change.
Please make sure to update tests as appropriate.
1.4k Dec 31, 2022
26 Aug 27, 2022
49 Dec 31, 2022
1 Feb 3, 2022
86 Dec 29, 2022
4 Aug 2, 2022
24 Jan 9, 2023
573 Dec 27, 2022
2.8k Jan 5, 2023
6 Oct 5, 2021
8 Mar 13, 2022
17 Jun 26, 2022
2 Jan 11, 2022
78 Jan 7, 2023
12 May 30, 2022
3 Feb 18, 2022
180 Jan 5, 2023
101 Nov 25, 2022
69 Dec 29, 2022
6 Oct 5, 2021
2 Jan 11, 2022
753 Dec 29, 2022
102 Jan 6, 2023
11 Nov 27, 2022
1 Jan 27, 2022
0 Oct 6, 2022
12.1k Jan 5, 2023
2k Dec 30, 2022