标签: python-asyncio

我目前正在编写一个脚本，该脚本首先使用 cfscrape 绕过 cloudflare，然后使用有效负载发出 2 个发布请求以登录该站点。我在 future1 和 future2 帖子中遇到一些错误。这是我的代码：

import asyncio
import requests
import cfscrape

async def main():
s = requests.Session()
s.get('https://www.off---white.com/en/IT')

headers = {
    'Referer': 'https://www.off---white.com/it/IT/login',
    'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/67.0.3396.99 Safari/537.36'
    }

payload1 = {
    'spree_user[email]': 'email',
    'spree_user[password]': 'password',
    'spree_user[remember_me]': '0',
}

payload2 = {
    'spree_user[email]': 'email',
    'spree_user[password]': 'password',
    'spree_user[remember_me]': '0',
}

scraper = cfscrape.create_scraper(s)
scraper.get('https://www.off---white.com/en/IT', headers=headers)
print('Done')

loop = asyncio.get_event_loop()
print('Starting loop')

future1 = loop.run_in_executor(None, requests.post ,'https://www.off---white.com/it/IT/login', data=payload1, headers=headers) …

我的服务器类继承自BaseServer：

class BaseServer(object):

    def __init__(self, host, port):
        self.loop = asyncio.new_event_loop()
        asyncio.set_event_loop(self.loop)

        self.instance = asyncio.start_server(self.handle_connection, host = host, port = port)

    async def handle_connection(self, reader: StreamReader, writer: StreamWriter):
        pass

    def start(self):
        # wrapping coroutine into ensure_future to allow it to call from call_soon
        # wrapping into lambda to make it callable
        callback = asyncio.ensure_future(self.instance)
        self.loop.call_soon(lambda: callback)
        self.loop.run_forever()
        self.loop.close()

    def stop(self):
        self.loop.call_soon_threadsafe(self.loop.stop)

    @staticmethod
    def get_instance():
        return BaseServer(None, None)

我需要两个在自己的线程中运行的服务器来并行处理请求。但是当我尝试根据需要运行它们时，只有第一台服务器正在运行。下面是我如何运行它们：

if __name__ == '__main__':
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)

    async def run(): …

我正在尝试使用 python 3.7.2 和 aiobotocore 包执行多个 AWS Lambda 调用。这是我的代码。

import asyncio
import aiobotocore


async def invoke(payload, session):
    async with session.create_client('lambda', region_name='us-east-1') as client:
        return await client.invoke(FunctionName='MY_FUNCTION', Payload=payload)


def generate_invocations(payloads, session):
    for payload in payloads:
        yield invoke(payload, session)


def invoke_all(payloads):
    loop = asyncio.get_event_loop()

    async def wrapped():
        session = aiobotocore.get_session(loop=loop)
        invocations = generate_invocations(payloads, session)
        return await asyncio.gather(*invocations)

    return loop.run_until_complete(wrapped())


def main():
    payloads_list = []  # MY PAYLOADS LIST 
    lambda_responses = invoke_all(payloads_list)
    print(lambda_responses)


if __name__ == '__main__':
    main()

代码运行速度非常快（10 个有效负载大约 1 秒，而不是使用 boto3 …

我正在使用 async io 对两种执行方法进行计时

情况1：

async def test():    
    print(f"started at {time.strftime('%X')}")
    await asyncio.create_task(say_after(2, 'hello'))
    await asyncio.create_task(say_after(4, 'world'))    
    print(f"finished at {time.strftime('%X')}")

它的回应是：

started at 12:31:05
hello
world
finished at 12:31:11

总共6秒

案例2：

async def test():    
    print(f"started at {time.strftime('%X')}")
    t1=asyncio.create_task(say_after(2, 'hello'))
    t2= asyncio.create_task(say_after(4, 'world'))    
    await t1
    await t2
    print(f"finished at {time.strftime('%X')}")

它的回应是：

started at 12:31:05
hello
world
finished at 12:31:09

总共4秒

为什么会这样呢？

我觉得我对异步 IO 的理解存在差距：在较大的协程范围内将小函数包装到协程中是否有好处？ 正确地发出事件循环信号有好处吗？这种好处的程度是否取决于包装的函数是 IO 还是 CPU 密集型？

示例：我有一个协程，download()其中：

1. 通过 . 从 HTTP 端点下载 JSON 序列化字节aiohttp。
2. 通过 - 压缩这些字节bz2.compress()- 这本身不是可等待的
3. 通过以下方式将压缩字节写入 S3aioboto3

因此，第 1 部分和第 3 部分使用这些库中的预定义协程；默认情况下，第 2 部分没有。

简化的例子：

import bz2
import io
import aiohttp
import aioboto3

async def download(endpoint, bucket_name, key):
    async with aiohttp.ClientSession() as session:
        async with session.request("GET", endpoint, raise_for_status=True) as resp:
            raw = await resp.read()  # payload (bytes)
            # Yikes - isn't it bad to throw a …

我正在运行一个程序，该程序从其余 api 发出三个不同的请求。数据、指标、请求函数都使用我制作的包装器从 BitMEX 的 api 获取数据。

我已经使用 asyncio 来尝试加快该过程，以便当我等待上一个请求的响应时，它可以开始发出另一个请求。

但是，由于某种原因，我的异步版本运行速度并没有更快。该代码有效，据我所知，我已正确设置所有内容。但我设置协程的方式可能有问题？

这是异步版本：

import time
import asyncio

from bordemwrapper import BitMEXData, BitMEXFunctions


'''
asynchronous I/O
'''

async def data():
    data = BitMEXData().get_ohlcv(symbol='XBTUSD', timeframe='1h', 
                                  instances=25)
    await asyncio.sleep(0)
    return data

async def indicator():
    indicator = BitMEXData().get_indicator(symbol='XBTUSD', 
        timeframe='1h', indicator='RSI', period=20, source='close', 
        instances=25)
    await asyncio.sleep(0)
    return indicator

async def request():
    request = BitMEXFunctions().get_price()
    await asyncio.sleep(0)
    return request

async def chain():
    data_ = await data()
    indicator_ = await indicator()
    request_ = await request()
    return data_, indicator_, …

我试图理解 python asyncio 的call_soon_threadsafe API，但失败了，使用下面的示例代码，如果我的simple协程想要返回某些内容，我应该如何从调用方获取返回值？

import time
import asyncio as aio
import uvloop

from threading import Thread

aio.set_event_loop_policy(uvloop.EventLoopPolicy())

async def simple(a, fut:aio.Future):
  await aio.sleep(a)
  return fut.set_result(a)

def delegator(loop):
  aio.set_event_loop(loop)
  loop.run_forever()

loop_exec = aio.new_event_loop()

t = Thread(target=delegator, args=(loop_exec,))
t.start()


if __name__ == '__main__':
  start_time = time.time()

  fut = loop_exec.create_future() # tried to get back returned value by future
  handle = loop_exec.call_soon_threadsafe(aio.ensure_future, simple(3, fut))
  res = aio.wait_for(fut, 10)

  print('Time consumed: {}s'.format(time.time() - start_time))
  print('>>>>>>>>>>', res)

# Output
Time consumed: …

我正在编写 python 上下文管理器，它运行异步任务。如果我的经理的任何任务抛出异常，我希望我的经理终止。这是示例代码：

class MyClass:
  def __init__(self):
    if asyncio.get_event_loop().is_closed():
      asyncio.set_event_loop(asyncio.new_event_loop())

    self.loop = asyncio.get_event_loop()

  def __enter__(self):
    return self

  def __exit__(self, excType, excValue, tb):
    try:
      self.loop.run_until_complete(self._exit_loop())
    finally:
      self.loop.close()

    if excType is not None:
      print(excType.__name__, ':', excValue)
      traceback.print_tb(tb)

  async def _exit_loop(self):
    tasks = [task for task in asyncio.all_tasks(self.loop) if
             task is not asyncio.current_task(self.loop)]
    list(map(lambda task: task.cancel(), tasks))

    results = await asyncio.gather(*tasks, return_exceptions=True)
    self.loop.stop()


  async def func1(self):
    while True:
      print('func1')
      await asyncio.sleep(1)

  async def func2(self):
    i = 5
    while i > 0:
      print('func2')
      await …

我正在尝试使用 python 包resuqests_html异步 get() URL 列表，类似于自述文件中使用 Python 3.6.5 和 requests_html 0.10.0 的异步示例。

我的理解是 AsyncHTMLSession.run() 应该与 asyncio.gather() 非常相似：你给它一堆可等待的东西，它会运行所有的东西。这是不正确的吗？

这是我正在尝试的代码，我希望它应该获取页面并存储响应：

from requests_html import AsyncHTMLSession

async def get_link(url):
    r = await asession.get(url)
    return r

asession = AsyncHTMLSession()
results = asession.run(get_link("http://google.com"), get_link("http://yahoo.com"))

但我却得到了这个异常：

Traceback (most recent call last):
  File "test.py", line 10, in <module>
    results = asession.run(get_link("google.com"), get_link("yahoo.com"))
  File ".\venv\lib\site-packages\requests_html.py", line 772, in run
    asyncio.ensure_future(coro()) for coro in coros
  File ".\venv\lib\site-packages\requests_html.py", line 772, in <listcomp>
    asyncio.ensure_future(coro()) for coro in coros
TypeError: …

我正在 Python 3.8 中使用 asyncio 库 https://docs.python.org/3/library/asyncio.html

我正在创建一个服务器，在“新接受的连接”回调函数中，我想找出新客户端的远程IP地址和端口。

回调函数的参数是 StreamReader 和 StreamWriter 各一个实例，用于从客户端读取和写入。有没有直接的方法来查找流的 IP 地址和端口？请注意，我想对 SSL 和非 SSL 连接执行此操作。

这里我创建服务器：

async def create_server(self, new_client_cb, host, port):
    srvsocket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    srvsocket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    srvsocket.bind((host, port))
    srvsocket.listen(5)
    return await asyncio.start_server(new_client_cb, sock=srvsocket, start_serving=False)

我传入回调函数，该函数遵循文档并接受 StreamReader 和 StreamWriter 的实例。

这里就是说回调函数。它是一个类的一部分，因此是主要的自我论证。

async def _new_client(self, client_r, client_w):
    try:
        self.logger.debug("New client on incoming proxy")
        dests_r = {}
        dests_w = {}
        for addr in self.config['addrlist']:
            host, port = addr.split(':')
            host = socket.gethostbyname(host)
            self.logger.debug(f"Connecting to {addr}...")
            r, w …