如何在 Python 中编写一系列 Promise？

Posted 2023-03-30

【中文标题】如何在 Python 中编写一系列 Promise？

【英文标题】：How do I write a sequence of promises in Python?

【发布时间】：2017-04-10 14:10:41

【问题描述】：

是否可以仅使用 Python 3.6.1 Standard Library 编写promise（或任务）序列？

例如，javascript 中的 promise 序列写成：

const SLEEP_INTERVAL_IN_MILLISECONDS = 200;

const alpha = function alpha (number) 
    return new Promise(function (resolve, reject) 
        const fulfill = function() 
            return resolve(number + 1);
        ;

        return setTimeout(fulfill, SLEEP_INTERVAL_IN_MILLISECONDS);
    );
;

const bravo = function bravo (number) 
    return new Promise(function (resolve, reject) 
        const fulfill = function() 
            return resolve(Math.ceil(1000*Math.random()) + number);
        ;
        return setTimeout(fulfill, SLEEP_INTERVAL_IN_MILLISECONDS);
    );
;

const charlie = function charlie (number) 
    return new Promise(function (resolve, reject) 
        return (number%2 == 0) ? reject(number) : resolve(number);
    );
;

function run() 
    return Promise.resolve(42)
        .then(alpha)
        .then(bravo)
        .then(charlie)
        .then((number) => 
            console.log('success: ' + number)
        )
        .catch((error) => 
            console.log('error: ' + error);
        );


run();

每个具有异步处理结果的函数also returns a Promise，将被紧随其后的promise解析/拒绝。

我知道promises-2.01b 和asyncio 3.4.3 等库，我正在寻找Python STL 解决方案。因此，如果我需要导入非 STL 库，我更喜欢使用 RxPython。

【参考方案1】：

这是一个使用 asyncio 和 async/await 语法的类似程序：

import asyncio
import random

async def alpha(x):
    await asyncio.sleep(0.2)
    return x + 1 

async def bravo(x):
    await asyncio.sleep(0.2)
    return random.randint(0, 1000) + x

async def charlie(x):
    if x % 2 == 0:
        return x
    raise ValueError(x, 'is odd')

async def run():
    try:
        number = await charlie(await bravo(await alpha(42)))
    except ValueError as exc:
        print('error:', exc.args[0])
    else:
        print('success:', number)

if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    loop.run_until_complete(run())
    loop.close()

---

编辑：如果您对反应式流感兴趣，可以考虑使用aiostream。

这是一个简单的例子：

import asyncio
from aiostream import stream, pipe

async def main():
    # This stream computes 11² + 13² in 1.5 second
    xs = (
        stream.count(interval=0.1)      # Count from zero every 0.1 s
        | pipe.skip(10)                 # Skip the first 10 numbers
        | pipe.take(5)                  # Take the following 5
        | pipe.filter(lambda x: x % 2)  # Keep odd numbers
        | pipe.map(lambda x: x ** 2)    # Square the results
        | pipe.accumulate()             # Add the numbers together
    )
    print('11² + 13² = ', await xs)

if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    loop.run_until_complete(main())
    loop.close()

documentation 中有更多示例。

_{免责声明：我是项目维护者。}

【讨论】：

酷！唯一不好的部分是伸缩方法调用。就像希望编写更像 JS 承诺序列或反应式可观察流的东西。但是这个已经解决了我的问题:-)

@JPVentura 是的，这一行有点长，但您可以使用额外的变量将其拆分。 asyncio 没有管道语法，因为它试图尽可能接近同步编程另外，请参阅我关于反应流的编辑。

@Vincent，你能看看这里吗：***.com/questions/53466252/…

除了害怕尝试新事物或懒得学习（这对开发人员来说很疯狂，但我一直都看到），你怎么能不喜欢函数式流管道......

【参考方案2】：

您很幸运，Python 3.4 及更高版本包括 asyncio，尽管您正在寻找的功能 (Future) 在 Python 3.5 及更高版本中可用。

来自您自己关于 asyncio 的链接：“此版本仅与 Python 3.3 相关，其标准库中不包含 asyncio。”

例子：

import asyncio


async def some_coroutine():
    await asyncio.sleep(1)
    return 'done'


def process_result(future):
    print('Task returned:', future.result())


loop = asyncio.get_event_loop()
task = loop.create_task(some_coroutine())
task.add_done_callback(process_result)
loop.run_until_complete()

【讨论】：

如何使用 Future 将前面的 JS 示例转换为 Python 3.5？

【参考方案3】：

您可以创建自己的 Class Promise，我不是 python 开发人员，但我尝试创建类似于 javascript 的东西。

class Promise:
    def __init__(self, callback):
        self.resolved = ''
        self.rejected = ''
        callback(self.resolve, self.reject)

    def resolve(self, value):
        self.resolved = value

    def reject(self, value):
        self.rejected = value

    def then(self, callback):
        if not self.rejected:
            self.resolved = callback(self.resolved)
        return self

    def catch(self, callback):
        if self.rejected:
            self.rejected = callback(self.rejected)
        return self


def myPromise(resolve, reject):
    resolve(['Ana', 'Bia', 'Carlos', 'Daniel'])


def firstResolve(value):
    return value[0]


def secondResolve(value):
    print(value)


def firstReject(value):
    print('error:', value)


p = Promise(myPromise)
p.then(firstResolve).then(secondResolve).catch(firstReject)

Promise.all 示例

class Promise:
    def __init__(self, callback):
        self.resolved = ''
        self.rejected = ''
        if callable(callback):
            callback(self.resolve, self.reject)

    def resolve(self, value):
        self.resolved = value

    def reject(self, value):
        self.rejected = value

    def then(self, callback):
        if not self.rejected:
            self.resolved = callback(self.resolved)
        return self

    def catch(self, callback):
        if self.rejected:
            self.rejected = callback(self.rejected)
        return self

    def all(self, promises):
        resolvedArray = []
        rejectedArray = []
        for promise in promises:
            promise(self.resolve, self.reject)
            if self.resolved:
                resolvedArray += self.resolved
            if self.rejected:
                rejectedArray += self.rejected
                break
        self.resolved = resolvedArray
        self.rejected = rejectedArray
        return self


def myPromise1(resolve, reject):
    resolve(['Ana'])


def myPromise2(resolve, reject):
    resolve(['Bia'])


def myPromise3(resolve, reject):
    resolve(['Carlos'])


def myPromise4(resolve, reject):
    resolve(['Daniel'])


def allResolve(values):
    print('without error: ', values)


def allReject(values):
    print('with error: ', values)


p = Promise([])
p.all([myPromise1, myPromise2]).then(allResolve).catch(allReject)

【讨论】：

有趣！但是对“回调”的调用是异步的吗？