zoukankan      html  css  js  c++  java
  • aiohttp

    发起请求

    async def fetch():
        async with aiohttp.ClientSession() as session:
            async with session.get('https://www.baidu.com') as resposne:
                print(await resposne.text())
    
    loop = asyncio.get_event_loop()
    tasks = [fetch(),]
    loop.run_until_complete(asyncio.wait(tasks))
    

    添加请求参数

    params = {'key': 'value', 'page': 10}
    async def fetch():
        async with aiohttp.ClientSession() as session:
            async with session.get('https://www.baidu.com/s',params=params) as resposne:
                print(await resposne.url)
    
    loop = asyncio.get_event_loop()
    tasks = [fetch(),]
    loop.run_until_complete(asyncio.wait(tasks))
    

    自定义User-Agent

    url = 'http://httpbin.org/user-agent'
    headers = {'User-Agent': 'test_user_agent'}
    
    async def fetch():
        async with aiohttp.ClientSession() as session:
            async with session.get(url,headers=headers) as resposne:
                print(await resposne.text())
    
    loop = asyncio.get_event_loop()
    tasks = [fetch(),]
    loop.run_until_complete(asyncio.wait(tasks))
    

    自定义cookies

    url = 'http://httpbin.org/cookies'
    cookies = {'cookies_name': 'test_cookies'}
    
    async def fetch():
        async with aiohttp.ClientSession() as session:
            async with session.get(url,cookies=cookies) as resposne:
                print(await resposne.text())
              
    
    loop = asyncio.get_event_loop()
    tasks = [fetch(),]
    loop.run_until_complete(asyncio.wait(tasks))
    

    post字符串

    url = 'http://httpbin.org'
    payload = {'username': 'zhang', 'password': '123456'}
    async def fetch():
        async with aiohttp.ClientSession() as session:
            async with session.post(url, data=payload) as resposne:
                print(await resposne.text())
    
    loop = asyncio.get_event_loop()
    tasks = [fetch(), ]
    loop.run_until_complete(asyncio.wait(tasks))
    

    post文件

    url = 'http://httpbin.org'
    files = {'file': open('test.txt', 'rb')}
    async def fetch():
        async with aiohttp.ClientSession() as session:
            async with session.post(url, data=files) as resposne:
                print(await resposne.text())
    
    loop = asyncio.get_event_loop()
    tasks = [fetch(), ]
    loop.run_until_complete(asyncio.wait(tasks))
    

    设置代理

    url = "http://python.org"
    async def fetch():
        async with aiohttp.ClientSession() as session:
            async with session.get(url, proxy="http://some.proxy.com") as resposne:
            print(resposne.status)
    
    loop = asyncio.get_event_loop()
    tasks = [fetch(), ]
    loop.run_until_complete(asyncio.wait(tasks))
    

    设置认证代理

    url = "http://python.org"
    async def fetch():
        async with aiohttp.ClientSession() as session:
            proxy_auth = aiohttp.BasicAuth('user', 'pass')
            async with session.get(url, proxy="http://some.proxy.com", proxy_auth=proxy_auth) as resposne:
                print(response.status)
    
    loop = asyncio.get_event_loop()
    tasks = [fetch(), ]
    loop.run_until_complete(asyncio.wait(tasks))
    
    # 下面的方法也可以
    url = "http://python.org"
    async def fetch():
        async with aiohttp.ClientSession() as session:
            async with session.get(url, proxy="http://user:pass@some.proxy.com") as response:
                print(response.status)
    
    loop = asyncio.get_event_loop()
    tasks = [fetch(), ]
    loop.run_until_complete(asyncio.wait(tasks))
    
    
  • 相关阅读:
    [git] push.default is unset
    [daily][qemu][libvirt] 使用libvirt管理qemu
    [development] __attribute__((weak))是干嘛的
    [daily] 使用左右对比查看diff 格式的文件
    [daily] 主机间目录共享
    [daily][qemu][kvm] 使用virtfs在host与guest之间共享目录
    [development][thrift] RPC框架 thrift
    [daily] 像tcpdump一样监听unix domain socket
    [daily][cgroup] 使用cgroup限制进程的CPU占用
    [qemu] qemu旧的net参数已经不再可用了,新的这样用。
  • 原文地址:https://www.cnblogs.com/cnkai/p/7642771.html
Copyright © 2011-2022 走看看