Concurrent map()

concurrent_map.py

from __future__ import annotations

import concurrent.futures
from typing import Callable, Iterable


def concurrent_map(func: Callable[[T], T2], *it: Iterable[T], max_n_threads: int) -> list[T2]:
    """
    Map a function to an iterable asynchronously

    Examples:

    >>> import time, requests
    ... list_of_urls = ...
    ... print('N URLs:', len(list_of_urls))
    ... start = time.perf_counter()
    ... concurrent_map(requests.get, list_of_urls, max_n_threads=10)
    ... print('Time taken:', time.perf_counter() - start)
    N URLs: 100
    Time taken: 0.7726614000002883
    """
    with concurrent.futures.ThreadPoolExecutor(max_workers=max_n_threads) as executor:
        # it's necessary to exhaust the iterator to "re-raise" the exceptions that happened when executing the function
        return list(executor.map(func, *it))