投稿
  1. 首页
  2. python

【Python自学笔记】aiohttp异步请求自用示例（备忘）

code • 2022-4-11 • python • 阅读 58

【Python自学笔记】aiohttp异步请求自用示例（备忘）,第1张

文章目录
    • 客户端使用
      • 基础请求
      • 获取响应
      • 响应流式内容
      • POST请求
      • POST流式上传
      • 设置超时
      • 使用代理
    • asyncio.run()报错

  • 官方文档
  • 【Python自学笔记】asyncio异步自用示例（备忘）
客户端使用 基础请求
  • 基本的url请求代码（和requests模块大致一样）
import aiohttp
import asyncio


async def get_response():
    async with aiohttp.ClientSession() as session:
        async with session.get('http://httpbin.org/get') as resp:
            print(resp.status)
            print(await resp.text())


asyncio.run(get_response())
  • 如果您访问的多个网站是在同一个根网站下（url链接前半部分都一样），可以使用构造函数的参数，如下代码
async with aiohttp.ClientSession('http://httpbin.org') as session:
    async with session.get('/get'):
        pass
    async with session.post('/post', data=b'data'):
        pass
    async with session.put('/put', data=b'data'):
        pass
获取响应
  • 我们在 requests模块 中使用最多的是r.content.decode()
  • 但是在这里，如下
import aiohttp
import asyncio


async def get_response():
    async with aiohttp.ClientSession() as session:
        async with session.get('http://httpbin.org/get') as resp:
            # 获取响应代码
            print(resp.status)

            # 获取解码后的文本内容，aiohttp自动解码来自服务器的内容。


            # 您也可以为该text(encoding='gbk')方法指定自定义编码：
            print(await resp.text())

            # 获取二进制内容（一般对于非文本请求使用）
            print(await resp.read())

            # 获取json数据，但是需要响应支持：如果 JSON 解码失败，json()将引发异常。

可以为json()调用指定自定义编码和解码器函数
            print(await resp.json())


asyncio.run(get_response())
响应流式内容
  • 如果下载一个图片或者引用
import aiohttp
import asyncio


async def get_response(url, filename='girl.png', chunk_size=1024):
    headers = {
        'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/80.0.3987.149 Safari/537.36'
    }
    async with aiohttp.ClientSession() as session:
        async with session.get(
                url,
                headers=headers) as resp:
            # 获取流式数据
            with open(filename, 'wb') as f:
                async for chunk in resp.content.iter_chunked(chunk_size):
                    f.write(chunk)
            print('下载完成...')


url = 'https://gimg2.baidu.com/image_search/src=http%3A%2F%2Fimg.jj20.com%2Fup%2Fallimg%2F1113%2F042620102J9%2F200426102J9-9-1200.jpg&refer=http%3A%2F%2Fimg.jj20.com&app=2002&size=f9999,10000&q=a80&n=0&g=0n&fmt=auto?sec=1652083513&t=3131e6dd818635093ace4df3feaeb711'
# asyncio.run(get_response())   # 这种形式运行会报错，因为这个函数是一个协程，不是一个线程
# 使用这种方法
loop = asyncio.get_event_loop()
loop.run_until_complete(get_response(url))
POST请求
  • post请求基本代码
import aiohttp
import asyncio


async def get_response(url):
    headers = {
        'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/80.0.3987.149 Safari/537.36'
    }
    async with aiohttp.ClientSession() as session:
        payload = {'key1': 'value1', 'key2': 'value2'}
        async with session.post(url, headers=headers, data=payload) as resp:

        # # 发送 bytes 数据
        # async with session.post(url, data=b'\x00Binary-data\x00') as resp:

        # #发送 JSON 数据：
        # async with session.post(url, json={'example': 'test'}) as resp:

        # # 发送 文本 数据
        # async with session.post(url, data='Text') as resp:

            print(await resp.text())


url = 'http://httpbin.org/post'
# asyncio.run(get_response(url))   # 这种形式运行结束后会报错，因为这个函数是一个协程，不是一个线程
# 使用这种方法
loop = asyncio.get_event_loop()
loop.run_until_complete(get_response(url))
POST流式上传
  • 当我们需要先服务器上传内容时候可以用以下代码
  1. 文件不是很大
import aiohttp
import asyncio


async def get_response(url, filename):
    headers = {
        'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/80.0.3987.149 Safari/537.36'
    }
    async with aiohttp.ClientSession() as session:
        with open(filename, 'rb') as f:
            await session.post(url, headers=headers, data=f)


filename = 'test.jpg'
url = 'http://httpbin.org/post'
# asyncio.run(get_response(url))   # 这种形式运行结束后会报错，因为这个函数是一个协程，不是一个线程
# 使用这种方法
loop = asyncio.get_event_loop()
loop.run_until_complete(get_response(url, filename))
  1. 文件很大，使用流式上传
import aiofiles as aiofiles
import aiohttp
import asyncio


# 异步生成器
async def file_sender(file_name=None):
    async with aiofiles.open(file_name, 'rb') as f:
        chunk = await f.read(64 * 1024)
        while chunk:
            yield chunk
            chunk = await f.read(64 * 1024)


async def main(url, filename):
    headers = {
        'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/80.0.3987.149 Safari/537.36'
    }
    async with aiohttp.ClientSession() as session:
        async with session.post(url, headers=headers, data=file_sender(file_name=filename))as resp:
            print(await resp.text())


filename = 'girl.png'
url = 'http://httpbin.org/post'
# asyncio.run(main(url))   # 这种形式运行结束后会报错，因为这个函数是一个协程，不是一个线程
# 使用这种方法
loop = asyncio.get_event_loop()
loop.run_until_complete(main(url, filename))
设置超时

默认情况下，aiohttp的超时时间为：300 秒（5 分钟），为了效率，我们需要进行自定义设置

  • 在requests模块中，我们都是直接使用timeout进行设置，但是aiohttp需要用到ClientTimeout，如下（注意有一个优先级问题在示例中被体现）
import aiohttp
import asyncio


async def get_response():
    # 设置超时，单位秒
    timeout = aiohttp.ClientTimeout(total=0.1)
    async with aiohttp.ClientSession(timeout=timeout) as session:
        timeout = aiohttp.ClientTimeout(total=1)
        # 注意如果2个地方都设置了超时，session.get()的超时优先级高于前面的timeout
        async with session.get('http://httpbin.org/get', timeout=timeout) as resp:
            print(resp.status)
            print(await resp.text())


# asyncio.run(get_response(url))   # 这种形式运行结束后会报错，因为这个函数是一个协程，不是一个线程
# 使用这种方法
loop = asyncio.get_event_loop()
loop.run_until_complete(get_response())
使用代理
  • 示例代码
import aiohttp
import asyncio


async def get_response(url, proxy):
    async with aiohttp.ClientSession() as session:
        # 设置代理
        async with session.get(url, proxy=proxy) as resp:
            print(resp.status)


url = 'http://httpbin.org/get'
proxy = 'http://192.168.1.88:5566'
# asyncio.run(get_response(url))   # 这种形式运行结束后会报错，因为这个函数是一个协程，不是一个线程
# 使用这种方法
loop = asyncio.get_event_loop()
loop.run_until_complete(get_response(url, proxy))
  • 官方说明
asyncio.run()报错

在较新的Python版本中我们都喜欢使用asyncio.run()运行，但是程序执行完成之后经常会报RuntimeError: Event loop is closed的错，我们可以修改一下写法，如下

asyncio.run(main())	# 有时候这样写会报错

# 这样写就OK了
loop = asyncio.get_event_loop()
loop.run_until_complete(main())
  • 报错原因：asyncio.run()会自动关闭循环,并且调用_ProactorBasePipeTransport.__del__报错, 而asyncio.run_until_complete()不会
  • 参考文章
  • 修改源码的解决方案

欢迎分享，转载请注明来源：内存溢出

原文地址: http://outofmemory.cn/langs/580182.html

赞 (0)
打赏 微信扫一扫 微信扫一扫 支付宝扫一扫 支付宝扫一扫
code code 管理员组
0 0
上一篇 2022-04-11
下一篇 2022-04-11

发表评论

登录后才能评论

评论列表（0条）

保存