首页 > 脚本中心 > Python > Python中aiohttp的简单使用

Python中aiohttp的简单使用

福牛 3月 28, 2023 27 0

1.定义

aiohttp 是一个基于 asyncio 的异步 HTTP 网络模块，它既提供了服务端，又提供了客户端

2.基本使用

  1. import aiohttp
  2. import asyncio
  3.  
  4.  
  5. async def fetch(session, url):
  6.      # 声明一个支持异步的上下文管理器
  7.      async with session.get(url) as response:
  8.      # response.text()是coroutine对象 需要加await
  9.      return await response.text(), response.status
  10.  
  11.  
  12. async def main():
  13.      # 声明一个支持异步的上下文管理器
  14.      async with aiohttp.ClientSession() as session:
  15.      html, status = await fetch(session, ‘https://cuiqingcai.com’)
  16.      print(f‘html: {html[:100]}…’)
  17.      print(f‘status: {status}’)
  18.  
  19.  
  20. if __name__ == ‘__main__’:
  21.      # python 3.7 及以后,不需要显式声明事件循环,可以使用 asyncio.run(main())来代替最后的启动操作
  22.      asyncio.get_event_loop().run_until_complete(main())

3.请求类型

  1. session.post(‘http://httpbin.org/post’, data=b‘data’)
  2. session.put(‘http://httpbin.org/put’, data=b‘data’)
  3. session.delete(‘http://httpbin.org/delete’)
  4. session.head(‘http://httpbin.org/get’)
  5. session.options(‘http://httpbin.org/get’)
  6. session.patch(‘http://httpbin.org/patch’, data=b‘data’)

4.相应字段

  1. print(‘status:’, response.status) # 状态码
  2. print(‘headers:’, response.headers)# 响应头
  3. print(‘body:’, await response.text())# 响应体
  4. print(‘bytes:’, await response.read())# 响应体二进制内容
  5. print(‘json:’, await response.json())# 响应体json数据

5.超时设置

  1. import aiohttp
  2. import asyncio
  3. async def main():
  4.      #设置 1 秒的超时
  5.      timeout = aiohttp.ClientTimeout(total=1)
  6.      async with aiohttp.ClientSession(timeout=timeout) as session:
  7.          async with session.get(‘https://httpbin.org/get’) as response:
  8.              print(‘status:’, response.status)
  9. if __name__ == ‘__main__’:
  10.      asyncio.get_event_loop().run_until_complete(main())

6.并发限制

  1. import asyncio
  2. import aiohttp
  3. # 声明最大并发量为5
  4. CONCURRENCY = 5
  5. semaphore = asyncio.Semaphore(CONCURRENCY)
  6. URL = ‘https://www.baidu.com’
  7.  
  8. session = None
  9. async def scrape_api():
  10.      async with semaphore:
  11.      print(‘scraping’, URL)
  12.      async with session.get(URL) as response:
  13.          await asyncio.sleep(1)
  14.          return await response.text()
  16. async def main():
  17.      global session
  18.      session = aiohttp.ClientSession()
  19.      scrape_index_tasks = [asyncio.ensure_future(scrape_api()) for _ in range(10000)]
  20.      await asyncio.gather(*scrape_index_tasks)
  21. if __name__ == ‘__main__’:
  22.      asyncio.get_event_loop().run_until_complete(main())

7.实际应用

  1. import asyncio
  2. import aiohttp
  3. import logging
  4. import json
  5. logging.basicConfig(level=logging.INFO,
  6.              format=‘%(asctime)s – %(levelname)s: %(message)s’)
  7. INDEX_URL = ‘https://dynamic5.scrape.center/api/book/?limit=18&offset={offset}’
  8. DETAIL_URL = ‘https://dynamic5.scrape.center/api/book/{id}’
  9. PAGE_SIZE = 18
  10. PAGE_NUMBER = 100
  11. CONCURRENCY = 5
  12.  
  13. semaphore = asyncio.Semaphore(CONCURRENCY)
  14. session = None
  15.  
  16. async def scrape_api(url):
  17.      async with semaphore:
  18.      try:
  19.          logging.info(‘scraping %s’, url)
  20.          async with session.get(url) as response:
  21.          return await response.json()
  22.      except aiohttp.ClientError:
  23.          logging.error(‘error occurred while scraping %s’, url, exc_info=True)
  24.  
  25. async def scrape_index(page):
  26.      url = INDEX_URL.format(offset=PAGE_SIZE * (page – 1))
  27.      return await scrape_api(url)
  28.  
  29. async def main():
  30.      global session
  31.      session = aiohttp.ClientSession()
  32.      scrape_index_tasks = [asyncio.ensure_future(scrape_index(page)) for page in range(1, PAGE_NUMBER + 1)]
  33.      results = await asyncio.gather(*scrape_index_tasks)
  34.      logging.info(‘results %s’, json.dumps(results, ensure_ascii=False, indent=2))
  36.  
  37. if __name__ == ‘__main__’:
  38.      asyncio.get_event_loop().run_until_complete(main())

到此这篇关于Python中aiohttp的简单使用的文章就介绍到这了,更多相关Python aiohttp 内容请搜索我们以前的文章或继续浏览下面的相关文章希望大家以后多多支持我们！

标签

相关推荐

发表评论