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
1.1k Dec 30, 2022
295 Jan 08, 2023
9.1k Jan 07, 2023
753 Dec 29, 2022
224 Nov 07, 2022
802 Dec 28, 2022
665 Jan 03, 2023
3.8k Jan 08, 2023
559 Dec 28, 2022
102 Jan 06, 2023
5k Jan 07, 2023
387 Dec 21, 2022
2.3k Jan 05, 2023
97 Oct 12, 2022