标签: python-asyncio

以下代码与TypeError: 'Mock' object is not iterablein 失败,ImBeingTested.i_call_other_coroutines因为我已被ImGoingToBeMockedMock对象替换.

如何模仿协同程序？

class ImGoingToBeMocked:
    @asyncio.coroutine
    def yeah_im_not_going_to_run(self):
        yield from asyncio.sleep(1)
        return "sup"

class ImBeingTested:
    def __init__(self, hidude):
        self.hidude = hidude

    @asyncio.coroutine
    def i_call_other_coroutines(self):
        return (yield from self.hidude.yeah_im_not_going_to_run())

class TestImBeingTested(unittest.TestCase):

    def test_i_call_other_coroutines(self):
        mocked = Mock(ImGoingToBeMocked)
        ibt = ImBeingTested(mocked)

        ret = asyncio.get_event_loop().run_until_complete(ibt.i_call_other_coroutines())

我正在编写一个用于枚举网站域名的python程序.例如,'a.google.com'.

首先,我使用该threading模块执行此操作:

import string
import time
import socket
import threading
from threading import Thread
from queue import Queue

'''
enumerate a site's domain name like this:
1-9 a-z + .google.com
1.google.com
2.google.com
.
.
1a.google.com
.
.
zz.google.com

'''

start = time.time()
def create_host(char):
    '''
    if char is '1-9a-z'
    create char like'1,2,3,...,zz'
    '''
    for i in char:
        yield i
    for i in create_host(char):
        if len(i)>1:
            return False
        for c in char:
            yield c + i


char = string.digits …

我正在尝试用Pygame和asyncio编写一个网络游戏,但我无法弄清楚如何避免挂起读取.这是我的客户代码:

@asyncio.coroutine
def handle_client():
    print("Connected!")
    reader, writer = yield from asyncio.open_connection('localhost', 8000)
    while True:
        mouse_up = False
        for event in pygame.event.get():
            if event.type == pygame.QUIT:
                pygame.quit()
                sys.exit()                
            elif event.type == pygame.MOUSEBUTTONUP:
                mouse_up = True

        if mouse_up:
            print("Writing")
            writer.write(b"Mouse up")
        print("Waiting to read")
        line = yield from reader.read(2**12)
        print(line.decode())

    writer.close()

这就行了line = yield from reader.read(2**12).我以前认为asyncio的意思是它是非阻塞的,所以如果没有任何数据可以读取它就会继续执行.我现在看到情况并非如此.

如何将asyncio网络代码与Pygame绘图和事件代码集成？

为了防止上下文切换,我想创建一个大循环来同时服务于网络连接和一些例程.

这是正常功能的实现:

import asyncio
import time


def hello_world(loop):
    print('Hello World')
    loop.call_later(1, hello_world, loop)

def good_evening(loop):
    print('Good Evening')
    loop.call_later(1, good_evening, loop)

print('step: asyncio.get_event_loop()')
loop = asyncio.get_event_loop()

print('step: loop.call_soon(hello_world, loop)')
loop.call_soon(hello_world, loop)
print('step: loop.call_soon(good_evening, loop)')
loop.call_soon(good_evening, loop)

try:
    # Blocking call interrupted by loop.stop()
    print('step: loop.run_forever()')
    loop.run_forever()
except KeyboardInterrupt:
    pass
finally:
    print('step: loop.close()')
    loop.close()

这是协同程序的实现:

import asyncio


@asyncio.coroutine
def hello_world():
    while True:
        yield from asyncio.sleep(1)
        print('Hello World')

@asyncio.coroutine
def good_evening():
    while True:
        yield from asyncio.sleep(1)
        print('Good Evening')

print('step: asyncio.get_event_loop()')
loop = asyncio.get_event_loop() …

我正在使用asyncioPython 3.5中的使用流程,但我还没有看到我应该做什么await以及我不应该做的事情或者可以忽略不计的事情的描述.我是否必须在"这是一个IO操作,因此应该await编辑" 方面使用我的最佳判断？

我想通过asyncio和连接到websocket websockets,格式如下所示.我怎么能做到这一点？

from websockets import connect


class EchoWebsocket:

    def __init__(self):
        self.websocket = self._connect()

    def _connect(self):
        return connect("wss://echo.websocket.org")

    def send(self, message):
        self.websocket.send(message)

    def receive(self):
        return self.websocket.recv()

echo = EchoWebsocket()
echo.send("Hello!")
print(echo.receive())  # "Hello!"

我有很少的阻止功能foo,bar我无法改变那些(一些我无法控制的内部库.与一个或多个网络服务交谈).我如何将其用作异步？我不想做以下事情.

results = []
for inp in inps:
    val = foo(inp)
    result = bar(val)
    results.append(result)

这将是低效的,因为我foo在等待第一个输入时可以调用第二个输入,并且相同bar.如何包装它们,这样它们与ASYNCIO(即新的使用async,await语法)？

让我们假设这些函数是可重入的.即,foo当先前foo正在处理时再次调用是可以的.

更新

用可重复使用的装饰器扩展答案.点击这里举例.

def run_in_executor(f):
    @functools.wraps(f)
    def inner(*args, **kwargs):
        loop = asyncio.get_running_loop()
        return loop.run_in_executor(None, functools.partial(f, *args, **kwargs))

    return inner

是否可以仅使用Python 3.6.1标准库编写一系列承诺(或任务)？

例如,JavaScript中的序列承诺写为:

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, …

从Python 3.5开始,关键字await和async语言都被引入.现在,我更像是一个Python 2.7人,而且我已经在相当长一段时间内避免使用Python 3了,所以asyncio对我来说这是一个新手.根据我的理解,它似乎await/async与它们在ES6(或JavaScript,ES2015中的工作方式)的工作方式非常相似,但是您想要称之为.

这是我用来比较它们的两个脚本.

import asyncio

async def countdown(n):
    while n > 0:
        print(n)
        n -= 1
        await asyncio.sleep(1)

async def main():
    """Main, executed in an event loop"""

    # Creates two countdowns
    futures = asyncio.gather(
        countdown(3), 
        countdown(2)
    )
    # Wait for all of them to finish
    await futures
    # Exit the app
    loop.stop()

loop = asyncio.get_event_loop()
asyncio.ensure_future(main())
loop.run_forever()

function sleep(n){
    // ES6 does not provide native sleep method with promise support …

我正在尝试http://httpbin.org/uuid使用以下代码片段一次向服务器发送 100 个请求

from fastapi import FastAPI
from time import sleep
from time import time
import requests
import asyncio

app = FastAPI()

URL= "http://httpbin.org/uuid"


# @app.get("/")
async def main():
    r = requests.get(URL)
    # print(r.text)
    
    return r.text

async def task():
    tasks = [main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main()]
    # print(tasks)
    # input("stop")
    result = await asyncio.gather(*tasks)
    print (result)

@app.get('/')
def f():
    start = time()
    asyncio.run(task())
    print("time: ",time()-start)

我将 FastAPI 与 Asyncio 结合使用，以实现大约 3 秒或更短的最短时间，但使用上述方法我得到的总时间为 66 秒，超过一分钟。我还想保留main用于附加操作的功能r.text。我知道要实现如此短的时间，需要并发性，但我不确定我在这里犯了什么错误。