标签: aiohttp

我刚刚开始使用Python3.4中的asyncio库,并编写了一个小程序,试图同时获取50个网页.该程序在几百个请求之后以"太多打开文件"异常爆炸.

我认为我的fetch方法用'response.read_and_close()'方法调用关闭连接.

有什么想法在这里发生了什么？我是以正确的方式解决这个问题吗？

import asyncio
import aiohttp

@asyncio.coroutine
def fetch(url):
    response = yield from aiohttp.request('GET', url)
    response = yield from response.read_and_close()
    return response.decode('utf-8')

@asyncio.coroutine
def print_page(url):
    page = yield from fetch(url)
    # print(page)

@asyncio.coroutine
def process_batch_of_urls(round, urls):
  print("Round starting: %d" % round)
  coros = []
  for url in urls:
      coros.append(asyncio.Task(print_page(url)))
  yield from asyncio.gather(*coros)
  print("Round finished: %d" % round)

@asyncio.coroutine
def process_all():
  api_url = 'https://google.com'
  for i in range(10):
    urls = []
    for url in range(50):
      urls.append(api_url)
    yield from process_batch_of_urls(i, urls) …

我找不到aiohttp与登录页面结合的工作代码.目标很简单:使用用户名和密码进行基于表单的身份验证,我希望在随后的aiohttp async fetch调用中使用该cookie.

似乎整个Session概念在版本之间的aiohttp中发生了变化,所以我很好奇如何在最新版本中实现它.我不确定如何获取cookie一次,然后在异步事件中使用它.

我真的很想看到一个完整的例子,因为不幸的是我无法使用我在任何地方找到的片段.

我想这可能是一个开始,但我不确定,我当然不知道如何将所有内容连接到它(我还需要一个aiohttp.TCPConnector吗？) http://aiohttp.readthedocs.org/en/latest/ client_reference.html#aiohttp.client.ClientSession

使用mechanize在Python 2中我的非异步版本的示例(虽然我自然地使用Python 3进行asyncio等):

import mechanize
import urllib

class MyClass()
    def __init__(self):
        self.data = {'username' : 'me', 'password' : 'pw'}
        self.login_url = 'http://example.com/login'
        self.login()

    def call(self, url):
        request2 = mechanize.Request(url)
        self.cookie_jar.add_cookie_header(request2)
        response2 = mechanize.urlopen(request2).read()
        return response2    

    def login(self):
        request = mechanize.Request(self.login_url)
        # 'username' and 'password' keys are actually the name of the <input>
        logInfo = urllib.urlencode({'username' : self.data['username'], 
                                    'password' : self.data['password']})
        response = mechanize.urlopen(request, data = logInfo) …

我正在尝试为以下异步编写pytest,等待方法,但我无处可去.

   class UserDb(object):
    async def add_user_info(self,userInfo):
      return await self.post_route(route='users',json=userInfo)

    async def post_route(self,route=None,json=None,params=None):
      uri = self.uri + route if route else self.uri      
      async with self.client.post(uri,json=json,params=params) as resp:            
      assert resp.status == 200
      return await resp.json()

有人可以帮我弄这个吗？TIA

aiohttp入门文档提供了以下客户端示例：

async with aiohttp.ClientSession() as session:
    async with session.get('https://api.github.com/events') as resp:
        print(resp.status)
        print(await resp.text())

我无法理解何时response.status可用。我的理解是协程在await response.read()生产线上释放了控制权。在等待响应恢复之前如何访问状态？

作为一项学习练习，我试图修改aiohttp的快速入门示例，以使用单个ClientSession提取多个URL（文档建议通常应为每个应用程序创建一个ClientSession）。

import aiohttp
import asyncio

async def fetch(session, url):
  async with session.get(url) as response:
    return await response.text()

async def main(url, session):
  print(f"Starting '{url}'")
  html = await fetch(session, url)
  print(f"'{url}' done")

urls = (
  "https://python.org",
  "https://twitter.com",
  "https://tumblr.com",
  "https://example.com",
  "https://github.com",
)

loop = asyncio.get_event_loop()
session = aiohttp.ClientSession()
loop.run_until_complete(asyncio.gather(
  *(loop.create_task(main(url, session)) for url in urls)
))
# session.close()   <- this doesn't make a difference

但是，在协程之外创建ClientSession显然不是可行的方法：

？python 1_async.py
1_async.py:30：用户警告：在协程之外创建客户端会话是一个非常危险的想法
  会话= aiohttp.ClientSession（）
在协程之外创建客户端会话
client_session： 
开始'https://python.org'
开始“ https://twitter.com”
开始'https://tumblr.com'
开始“ https://example.com”
开始'https://github.com'
'https://twitter.com'完成
“ https://example.com”完成
'https://github.com'完成 …

我想以异步方式将多个对象从一个 s3 存储桶复制到另一个存储桶，并且我想保持其状态。

我已经尝试在 s3 boto3 库上使用 async 和 wait ，但它们执行顺序复制。我也尝试过使用 aiobotocore 和 aioboto3 库，但它们没有 boto3 中使用的正确的 copy_object API。

async def copy_file(taskdescriptor: list, buckets: list):
    #loop = asyncio.get_event_loop()
    print("taskdescriptor", taskdescriptor)
    for job in taskdescriptor[0]:
        print("Inside copy", job , buckets)

        src_bucket = buckets[0]
        dest_bucket = buckets[1]
        src_path = job[0] 
        dest_path = job[1]
        src_loc = {"Bucket": src_bucket, "Key": src_path}
        #loop = asyncio.get_event_loop()
        #session = aiobotocore.get_session()
        async with aioboto3.client('s3') as asyncs3:
            print("s3 copy from {} to {}".format(src_path, dest_path))
            resp = await asyncs3.copy_object(Bucket=dest_bucket, Key=dest_path, …

我正在使用 pytest 编写 python 测试，并且有一些我想测试的异步代码，因此我安装了 pytest-asyncio 插件。异步代码使用 aiohttp 并运行测试，在测试成功运行后，我收到以下警告/错误/提示。

========================================================================================================================================================================= test session starts =========================================================================================================================================================================
platform win32 -- Python 3.8.7, pytest-6.2.1, py-1.10.0, pluggy-0.13.1
rootdir: C:\Users\user\workspace\example\app
plugins: asyncio-0.14.0
collected 1 item                                                                                                                                                                                                                                                                                                                                                       

tests\test_something.py .

========================================================================================================================================================================== 1 passed in 1.22s ==========================================================================================================================================================================
Exception ignored in: <function _ProactorBasePipeTransport.__del__ at 0x000001D09DE4C310>
Traceback (most recent call last):
  File "C:\Users\user\AppData\Local\Programs\Python\Python38\lib\asyncio\proactor_events.py", line 116, in __del__
    self.close()
  File "C:\Users\user\AppData\Local\Programs\Python\Python38\lib\asyncio\proactor_events.py", line 108, in close
    self._loop.call_soon(self._call_connection_lost, None)
  File "C:\Users\user\AppData\Local\Programs\Python\Python38\lib\asyncio\base_events.py", line 719, in call_soon
    self._check_closed()
  File "C:\Users\user\AppData\Local\Programs\Python\Python38\lib\asyncio\base_events.py", line 508, in _check_closed
    raise RuntimeError('Event loop …

我正在使用 Aiohttp 的 multidict() 实现。拿着它：

>>> d = MultiDict[('a', 1), ('b', 2), ('a', 3)])
>>> d
<MultiDict {'a': 1, 'b': 2, 'a': 3}>

我想转换d为常规字典，其中重复的键值被附加到列表中，如下所示：

{'a': [1, 3], 'b': 2}

有没有一种优雅的方法来转换它？除了循环遍历项目和大量逻辑条件之外？

SelectorEventLoop和到底是什么ProactorEventLoop？它们有何不同？

我正在 Windows 上的 python 中测试asyncio和aiohttp的使用，然后出现错误RuntimeError: Event loop is closed。查找并找到了解决方法，但我不明白错误的根源。许多人似乎都触及了and here和 also here
的概念。SelectorEventLoopProactorEventLoop

我阅读了文档但无法理解这个概念。谁能用更简单的术语解释这些概念？

下面是一个收集 URL 长度的简单程序。

import aiohttp
import asyncio
from time import perf_counter


URLS = ['http://www.cnn.com', 'http://www.huffpost.com', 'http://europe.wsj.com',
        'http://www.bbc.co.uk', 'http://failfailfail.com']

async def async_load_url(url, session):
    try:
        async with session.get(url) as resp:
            content = await resp.read()
        print(f"{url!r} is {len(content)} bytes")
    except IOError:
        print(f"failed to load {url}")        

async def main():

    async with aiohttp.ClientSession() as session:
        tasks = [async_load_url(url, session) for url in URLS]
        await asyncio.wait(tasks)

if __name__ == "__main__":
    start = perf_counter()
    asyncio.run(main())
    elapsed = perf_counter() - start
    print(f"\nTook {elapsed} seconds")

为什么以下代码在 python 3.9 中失败并出现运行时错误并忽略异常？如何修复它？ …