Promises-like API in Python

conts.py

import multiprocessing

from collections import namedtuple
from concurrent.futures import ThreadPoolExecutor

Poll = namedtuple('Poll', ['resolved', 'value'])

_CURRENT_EXECUTOR = None

def enter_runtime(main=None, threads=0):
    if main == None:
        raise PromiseError('provide a `main` argument')

    if threads == 0:
        threads = multiprocessing.cpu_count()

    global _CURRENT_EXECUTOR
    with ThreadPoolExecutor(max_workers=threads) as executor:
        _CURRENT_EXECUTOR = executor
        main()

    _CURRENT_EXECUTOR = None

class PromiseError(Exception):
    ...

class Promise(object):
    def __init__(self, body):
        self._run(body)

    def resolve_with(value):
        return Promise(lambda resolve, _: resolve(value()))

    def resolve(x):
        return Promise.resolve_with(lambda: x)

    def reject_with(err):
        return Promise(lambda _, reject: reject(err()))

    def reject_with(err):
        return Promise.resolve_with(lambda: err)

    def then(self, fmap):
        def next(resolve, reject):
            # resolve future
            self._future.result()
            resolve(fmap(self._value))
        return Promise(next)

    def poll(self):
        if self._future.running():
            return Poll(False, None)
        else:
            return Poll(True, self._value)

    # suffix await syntax ftw
    @property
    def wait(self):
        self._future.result()
        return self._value

    def _run(self, body):
        def reject(err):
            raise PromiseError(err)

        def resolve(value):
            self._value = value

        def future_main():
            body(resolve, reject)

        global _CURRENT_EXECUTOR
        if not _CURRENT_EXECUTOR:
            raise PromiseError('no executor found')

        self._future = _CURRENT_EXECUTOR.submit(future_main)

def async_main():
    # using functor style
    functor_result = Promise.resolve(5) \
        .then(lambda x: x + 2) \
        .then(lambda x: x * 10) \
        .wait

    # using monadic style
    five = Promise.resolve_with(lambda: 5).wait
    seven = Promise.resolve_with(lambda: five + 2).wait
    monadic_result = Promise.resolve_with(lambda: seven * 10).wait

    assert functor_result == monadic_result
    print(functor_result)

if __name__ == '__main__':
    enter_runtime(main=async_main)