Functional interface for concurrent futures, including asynchronous I/O.

from futured import threaded, processed
import httpx

fetch = threaded(httpx.Client().get)
fetch(url)  # return Future

fs = (fetch(url + path) for path in paths)
threaded.results(fs)  # generate results from futures
threaded.results(fs, timeout=...)  # generate results as completed

fetch.map(urls)  # generate results in order
fetch.map(urls, timeout=...)  # generate results as completed
fetch.mapzip(urls)  # generate (url, result) pairs as completed

threaded(max_workers=...)(func, ...)
processed(max_workers=...)(func, ...)

with threaded.waiting(*fs) as tasks:
    tasks.append(future)
tasks  # list of completed results

threaded.tasks(fs, timeout=...)  # mutable set of running tasks which iterate as completed

from futured import asynced
import httpx

fetch = asynced(httpx.AsyncClient().get)
fetch(url)  # return coroutine

asynced.results(fs)  # generate results from futures
asynced.results(fs, timeout=...)  # generate results as completed

fetch.map(urls)  # generate results in order
fetch.map(urls, timeout=...)  # generate results as completed
fetch.mapzip(urls)  # generate (url, result) pairs as completed

asynced.run(async_func, ...)  # call and run until complete
asynced.run(async_gen, ...)  # call and run synchronous iterator
with asynced.waiting(*fs) as tasks:  # concurrent coroutines completed in a block
asynced.tasks(fs, timeout=...)  # mutable set of running tasks which iterate as completed

@threaded
def slow():
   ...

fetch = threaded(httpx.Client().get, url)
fetch(params=...)

from futured import decorated

FutureClient = decorated(httpx.Client, request=threaded)

 # equivalent to
class FutureClient(httpx.Client):
    request = threaded(httpx.Client.request)

from futured import futured, command

command('ls').result()  # return stdout or raises stderr
command('ls').pipe('wc')  # pipes into next command, or | ('wc',... )
for line in command('ls'):  # iterable lines
command.coroutine('ls')  # return coroutine

futured(command, 'ls')  # supports `map` interface
asynced(command.coroutine, 'ls')  # supports `map` interface with timeout

from futured import forked

for value in forked(values, max_workers=...):
    # in a child process
 # in parent after children have exited

% pip install futured

% pytest [--cov]