futured 1.0

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

Futured provides a simple consistent interface for concurrent functional programming in Python. It wraps any callable to return a concurrent.futures.Future, wraps any async coroutine to return an asyncio.Future, and provides concurrent iterators and context managers for futures.

Usage

threaded, processed

Transform any callable into one which runs in a thread or process pool, and returns a future.

from futured import threaded, processed
import requests

fetch = threaded(requests.Session().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

Thread and process pool executors may be used as context managers, customized with options, and reused with different callables.

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

futured classes have a waiting context manager which collects results from tasks. Futures can be registered at creation, or appended to the list of tasks.

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

asynced

The same interface works for asyncio.

from futured import asynced
import aiohttp

fetch = asynced(aiohttp.ClientSession().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 provides utilities for calling coroutines from a synchronous context. waiting is similar to trio's nursery, but returns results from a synchronous with block.

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

decorators

Naturally futured wrappers can be used as decorators, but arguments can also be partially bound.

@threaded
def slow():
   ...

fetch = threaded(requests.Session().get, url)
fetch(params=...)

Methods are supported, as well as a decorated utility for automatically subclassing.

from futured import decorated

FutureSession = decorated(requests.Session, request=threaded)

 # equivalent to
class FutureSession(requests.Session):
    request = threaded(requests.Session.request)

command

command wraps subprocess.Popen to provide a Future compatible interface.

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

forked

forked allows iteration in separate child processes.

from futured import forked

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

Installation

$ pip install futured

Tests

100% branch coverage.

$ pytest [--cov]

Changes

1.0

• Executed functions are context managers
• starmap supported

0.3

• forked has optional maximum number of workers
• waiting context manager
• command pipes (|)
• distributed.Client support

0.2

• command.coroutine creates asyncio subprocesses
• futured.mapzip generates results zipped with arguments
• asynced.run supports asynchronous iterators