1.requests请求

# -*- coding: utf-8 -*-
# @Time    : 2022/12/6 16:03
# @Author  : 红后
# @Email   : not_enabled@163.com
# @blog    : https://www.cnblogs.com/Red-Sun
# @File    : 实例1.py
# @Software: PyCharm
import aiohttp, asyncio


async def aiohttp_requests(url):  # aiohttp的requests函数
    async with aiohttp.request("GET", url=url) as response:
        return await response.text(encoding='UTF-8')


async def main():  # 主函数用于异步函数的启动
    url = 'https://www.baidu.com'
    html = await aiohttp_requests(url)  # await修饰异步函数
    print(html)


if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    loop.run_until_complete(main())

2.session请求

GET：

# -*- coding: utf-8 -*-
# @Time    : 2022/12/6 16:33
# @Author  : 红后
# @Email   : not_enabled@163.com
# @blog    : https://www.cnblogs.com/Red-Sun
# @File    : 实例2.py
# @Software: PyCharm
import aiohttp, asyncio


async def aiohttp_requests(url):  # aiohttp的requests函数
    async with aiohttp.ClientSession() as session:  # 声明了一个支持异步的上下文管理器
        async with session.get(url) as response:
            return await response.text(encoding='UTF-8')


async def main():  # 主函数用于异步函数的启动
    url = 'https://www.baidu.com'
    html = await aiohttp_requests(url)  # await修饰异步函数
    print(html)


if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    loop.run_until_complete(main())

其中aiohttp还有post，put, delete...等一系列请求（PS：一般情况下只需要创建一个session，然后使用这个session执行所有的请求。）
PSOT：传参

async def aiohttp_requests(url):  # aiohttp的requests函数
    async with aiohttp.ClientSession() as session:
        data = {'key': 'value'}
        async with session.post(url=url, data=data) as response:
            return await response.text(encoding='UTF-8')

PS：这种传参传递的数据将会被转码，如果不想被转码可以直接提交字符串data=str(data)

附：关于session请求数据修改操作

1.cookies

自定义cookies应该放在ClientSession中，而不是session.get()中

async def aiohttp_requests(url):  # aiohttp的requests函数
    cookies = {'USER_AGENT': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/106.0.0.0 Safari/537.36'}
    async with aiohttp.ClientSession(cookies=cookies) as session:
        async with session.get(url) as response:
            return await response.text(encoding='UTF-8')

2.headers

放在自定义的headers跟正常的requests一样放在session.get()中

async def aiohttp_requests(url):  # aiohttp的requests函数
    async with aiohttp.ClientSession() as session:
        headers = {'USER_AGENT': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/106.0.0.0 Safari/537.36'}
        async with session.get(url=url, headers=headers) as response:
            return await response.text(encoding='UTF-8')

3.timeout

默认响应时间为5分钟，通过timeout可以重新设定，其放在session.get()中

async def aiohttp_requests(url):  # aiohttp的requests函数
    async with aiohttp.ClientSession() as session:
        async with session.get(url=url, timeout=60) as response:
            return await response.text(encoding='UTF-8')

4.proxy

当然代理也是支持的在session.get()中配置

async def aiohttp_requests(url):  # aiohttp的requests函数
    async with aiohttp.ClientSession() as session:
        async with session.get(url=url, proxy="http://some.proxy.com") as response:
            return await response.text(encoding='UTF-8')

需要授权的代理

async def aiohttp_requests(url):  # aiohttp的requests函数
    async with aiohttp.ClientSession() as session:
        proxy_auth = aiohttp.BasicAuth('user', 'pass')  # 用户，密码
        async with session.get(url=url, proxy="http://some.proxy.com", proxy_auth=proxy_auth) as response:
            return await response.text(encoding='UTF-8')

或者

async def aiohttp_requests(url):  # aiohttp的requests函数
    async with aiohttp.ClientSession() as session:
        async with session.get(url=url, proxy='http://user:pass@some.proxy.com') as response:
            return await response.text(encoding='UTF-8')

报错处理

错误：RuntimeError: Event loop is closed
报错原因是使用了asyncio.run(main())来运行程序
看到别个大佬的总结是asyncio.run()会自动关闭循环,并且调用_ProactorBasePipeTransport.__del__报错, 而asyncio.run_until_complete()不会。
第一种解决方法换成如下代码运行

loop = asyncio.get_event_loop()
loop.run_until_complete(main())

第二种重写方法以保证run()的运行

from functools import wraps

from asyncio.proactor_events import _ProactorBasePipeTransport

def silence_event_loop_closed(func):
    @wraps(func)
    def wrapper(self, *args, **kwargs):
        try:
            return func(self, *args, **kwargs)
        except RuntimeError as e:
            if str(e) != 'Event loop is closed':
                raise
    return wrapper

_ProactorBasePipeTransport.__del__ = silence_event_loop_closed(_ProactorBasePipeTransport.__del__)

标签：Python,爬虫,零基础必,实战教程,数据,链接,数字字母 来源：

本站声明： 1. iCode9 技术分享网（下文简称本站）提供的所有内容，仅供技术学习、探讨和分享； 2. 关于本站的所有留言、评论、转载及引用，纯属内容发起人的个人观点，与本站观点和立场无关； 3. 关于本站的所有言论和文字，纯属内容发起人的个人观点，与本站观点和立场无关； 4. 本站文章均是网友提供，不完全保证技术分享内容的完整性、准确性、时效性、风险性和版权归属；如您发现该文章侵犯了您的权益，可联系我们第一时间进行删除； 5. 本站为非盈利性的个人网站，所有内容不会用来进行牟利，也不会利用任何形式的广告来间接获益，纯粹是为了广大技术爱好者提供技术内容和技术思想的分享性交流网站。