大型fastapi项目实战 高并发请求神器之aiohttp(上) [建议收藏]
在 Python 众多的 HTTP 客户端中,最有名的莫过于 requests、aiohttp 和 httpx。在不借助其他第三方库的情况下,requests 只能发送同步请求;aiohttp 只能发送异步请求;httpx 既能发送同步请求,又能发送异步请求。在并发量大的情况下,如何高效的处理数据,异步是我们的优选,今天我们主要详解的是在生产环境广泛使用的 aiohttp。
aiohttp 是一个为 Python 提供异步HTTP 客户端/服务端编程,基于 asyncio(Python用于支持异步编程的标准库)的异步库。
# -*- encoding: utf-8 -*-
import asyncio
import aiohttp
async def main():
async with aiohttp.ClientSession() as session:
async with session.get('http://www.baidu.com') as resp:
print(resp.status)
res = await resp.text()
print(res[:100])
if __name__ == '__main__':
# 注意:
# python3.7+ 支持写法
# asyncio.run(main())
# python3.6及以下版本写法
event_loop = asyncio.get_event_loop()
result = event_loop.run_until_complete(asyncio.gather(main()))
event_loop.close()
# -*- encoding: utf-8 -*-
import asyncio
import aiohttp
async def post_v1():
data = b'\x00Binary-data\x00' # 未经编码的数据通过bytes数据上传
data = 'text' # 传递文本数据
data = {'key': 'value'} # 传递form表单
async with aiohttp.ClientSession() as sess:
async with sess.post('http://httpbin.org/post', data=data) as resp:
print(resp.status)
# 复杂的 post 请求
async def post_v2():
payload = {'key': 'value'} # 传递 pyload
async with aiohttp.ClientSession() as sess:
async with sess.post('http://httpbin.org/post', json=payload) as resp:
print(resp.status)
if __name__ == '__main__':
event_loop = asyncio.get_event_loop()
result = event_loop.run_until_complete(asyncio.gather(main()))
event_loop.close()
有些场景是需要拼接请求url 在这个时候可以使用本 case 来做处理
# -*- encoding: utf-8 -*-
import asyncio
import aiohttp
async def main():
""" 以下三种方式均可以 """
params = {'key1': 'value1', 'key2': 'value2'}
params = [('key', 'value1'), ('key', 'value2')]
params = 'key=value+1'
async with aiohttp.ClientSession() as sess:
async with sess.get('http://httpbin.org/get', params=params) as resp:
print(resp.status)
if __name__ == '__main__':
event_loop = asyncio.get_event_loop()
result = event_loop.run_until_complete(asyncio.gather(main()))
event_loop.close()
有时候,我们确实是有想服务器传文件的需求,eg:上传回执单;上传图片...... 100张 10000张的量级的时候我们会想用多线程去处理,但量再大 你再使用 多线程+requests 的方式就会发现有大量的报错,若有类似的使用场景,可以用以下 case 处理
import aiohttp
async def main():
""" 传递文件 """
files = {'file': open('report.xls', 'rb')}
async with aiohttp.ClientSession() as sess:
async with sess.post('http://httpbin.org/post', data=files) as resp:
print(resp.status)
print(await resp.text())
async def main2():
""" 实例化 FormData 可以指定 filename 和 content_type """
data = aiohttp.FormData()
data.add_field('file',
open('report.xls', 'rb'),
filename='report.xls',
content_type='application/vnd.ms-excel')
async with aiohttp.ClientSession() as sess:
async with sess.post('http://httpbin.org/post', data=data) as resp:
print(resp.status)
print(await resp.text())
async def main3():
""" 流式上传文件 """
async with aiohttp.ClientSession() as sess:
with open('report.xls', 'rb') as f:
async with sess.post('http://httpbin.org/post', data=f) as resp:
print(resp.status)
print(await resp.text())
async def main4():
""" 因为 content属性是 StreamReader(提供异步迭代器协议),
所以可以将 get 和 post 请求链接在一起。python3.6+能使用"""
async with aiohttp.ClientSession() as sess:
async with sess.get('http://python.org') as resp:
async with sess.post('http://httpbin.org/post', data=resp.content) as r:
print(r.status)
print(await r.text())
有时候,我们向服务器发送请求,若没有设置超时时间,此请求就会一直阻塞直到系统报错,这对于我们的系统是无法容忍的,所以发请求的时候千万要记得加上超时时间。
import aiohttp
timeout = aiohttp.ClientTimeout(total=60)
async def main():
async with aiohttp.ClientSession(timeout=timeout) as sess:
async with sess.get('http://httpbin.org/get') as resp:
print(resp.status)
print(await resp.text())
import aiohttp
import asyncio
cookies = {'cookies_are': 'working'}
async def main():
async with aiohttp.ClientSession(cookies=cookies) as session:
async with session.get('http://httpbin.org/cookies') as resp:
print(resp.status)
print(await resp.text())
assert await resp.json() == {"cookies": {"cookies_are": "working"}}
if __name__ == "__main__":
event_loop = asyncio.get_event_loop()
result = event_loop.run_until_complete(asyncio.gather(main()))
event_loop.close()
import aiohttp
import asyncio
async def main():
async with aiohttp.ClientSession() as session:
await session.get('http://httpbin.org/cookies/set?my_cookie=my_value')
filtered = session.cookie_jar.filter_cookies('http://httpbin.org')
print(filtered)
assert filtered['my_cookie'].value == 'my_value'
async with session.get('http://httpbin.org/cookies') as r:
json_body = await r.json()
print(json_body)
assert json_body['cookies']['my_cookie'] == 'my_value'
if __name__ == "__main__":
event_loop = asyncio.get_event_loop()
result = event_loop.run_until_complete(asyncio.gather(main()))
event_loop.close()
Cookie 的安全性问题: 默认 ClientSession 使用的是严格模式的 aiohttp.CookieJar. RFC 2109,明确的禁止接受url和ip地址产生的 cookie,只能接受 DNS 解析IP产生的cookie。
可以通过设置 aiohttp.CookieJar 的 unsafe=True 来配置
jar = aiohttp.CookieJar(unsafe=True)
session = aiohttp.ClientSession(cookie_jar=jar)
使用虚假Cookie Jar: 有时不想处理cookie。这时可以在会话中使用aiohttp.DummyCookieJar来达到目的。
jar = aiohttp.DummyCookieJar()
session = aiohttp.ClientSession(cookie_jar=jar)
import aiohttp
import asyncio
async with aiohttp.ClientSession(headers={'User-Agent': 'your agent'
"refer":"http://httpbin.org"}) as session:
async with session.get('http://httpbin.org/headers') as resp:
print(resp.status)
print(await resp.text())
默认情况下,aiohttp对HTTPS协议使用严格检查,如果你不想上传SSL证书,可将ssl设置为False。
r = await session.get('https://example.com', ssl=False)
# 第一种
async with aiohttp.ClientSession() as session:
proxy_auth = aiohttp.BasicAuth('user', 'pass')
async with session.get("http://python.org", proxy="http://proxy.com", proxy_auth=proxy_auth) as resp:
print(resp.status)
# 第二种
session.get("http://python.org", proxy="http://user:pass@some.proxy.com")
想要调整请求的传输层你可以为ClientSession及其同类组件传递自定义的连接器。例如:
conn = aiohttp.TCPConnector()
session = aiohttp.ClientSession(connector=conn)
注:不要给多个会话对象使用同一个连接器,某一会话对象拥有其所有权。
限制同一时间打开的连接数可以传递limit参数:
conn = aiohttp.TCPConnector(limit=30)
这样就将总数限制在30,默认情况下是100.如果你不想有限制,传递0即可:
conn = aiohttp.TCPConnector(limit=0)
爬虫常用的功能单独来写,主要是 aiohttp 还有一个问题没有解决,通过阅读源码确实是无法很好解决这个问题,在网上搜索了大半天基本没有有效的解决方案,so 笔者会给出一个自己找到的解决方案,在接下来的文章中我会进行分享。
本文分享自 python编程从入门到实践 微信公众号,前往查看
如有侵权,请联系 cloudcommunity@tencent.com 删除。
本文参与 腾讯云自媒体同步曝光计划 ,欢迎热爱写作的你一起参与!