快捷導(dǎo)航

Python中aiohttp的簡單使用

更新時(shí)間：2023年03月28日 09:26:54 作者：小Pawn爺

aiohttp是Python中一個(gè)強(qiáng)大的異步HTTP客戶端和服務(wù)器框架，它可以幫助開發(fā)者快速構(gòu)建高性能的Web應(yīng)用程序。本文將介紹aiohttp的基本概念、使用方法和常見應(yīng)用場景，幫助讀者更好地了解和使用這個(gè)優(yōu)秀的框架

1.定義

aiohttp 是一個(gè)基于 asyncio 的異步 HTTP 網(wǎng)絡(luò)模塊，它既提供了服務(wù)端，又提供了客戶端

2.基本使用

import aiohttp
import asyncio


async def fetch(session, url):
? ? # 聲明一個(gè)支持異步的上下文管理器
? ? async with session.get(url) as response:
? ? ? ? # response.text()是coroutine對象 需要加await
? ? ? ? return await response.text(), response.status


async def main():
? ? # 聲明一個(gè)支持異步的上下文管理器
? ? async with aiohttp.ClientSession() as session:
? ? ? ? html, status = await fetch(session, 'https://cuiqingcai.com')
? ? ? ? print(f'html: {html[:100]}...')
? ? ? ? print(f'status: {status}')


if __name__ == '__main__':
? ? # ?Python 3.7 及以后,不需要顯式聲明事件循環(huán),可以使用 asyncio.run(main())來代替最后的啟動(dòng)操作
? ? asyncio.get_event_loop().run_until_complete(main())

3.請求類型

session.post('http://httpbin.org/post', data=b'data')
session.put('http://httpbin.org/put', data=b'data')
session.delete('http://httpbin.org/delete')
session.head('http://httpbin.org/get')
session.options('http://httpbin.org/get')
session.patch('http://httpbin.org/patch', data=b'data')

4.相應(yīng)字段

print('status:', response.status) # 狀態(tài)碼
print('headers:', response.headers)# 響應(yīng)頭
print('body:', await response.text())# 響應(yīng)體
print('bytes:', await response.read())# 響應(yīng)體二進(jìn)制內(nèi)容
print('json:', await response.json())# 響應(yīng)體json數(shù)據(jù)

5.超時(shí)設(shè)置

import aiohttp
import asyncio
async def main():
   #設(shè)置 1 秒的超時(shí) 
   timeout = aiohttp.ClientTimeout(total=1)
   async with aiohttp.ClientSession(timeout=timeout) as session:
       async with session.get('https://httpbin.org/get') as response:
           print('status:', response.status)
if __name__ == '__main__':
   asyncio.get_event_loop().run_until_complete(main())

6.并發(fā)限制

import asyncio
import aiohttp
# 聲明最大并發(fā)量為5
CONCURRENCY = 5
semaphore = asyncio.Semaphore(CONCURRENCY)
URL = 'https://www.baidu.com'

session = None
async def scrape_api():
? ?async with semaphore:
? ? ? ?print('scraping', URL)
? ? ? ?async with session.get(URL) as response:
? ? ? ? ? ?await asyncio.sleep(1)
? ? ? ? ? ?return await response.text()
? ??
async def main():
? ?global session
? ?session = aiohttp.ClientSession()
? ?scrape_index_tasks = [asyncio.ensure_future(scrape_api()) for _ in range(10000)]
? ?await asyncio.gather(*scrape_index_tasks)
if __name__ == '__main__':
? ?asyncio.get_event_loop().run_until_complete(main())

7.實(shí)際應(yīng)用

import asyncio
import aiohttp
import logging
import json
logging.basicConfig(level=logging.INFO,
? ? ? ? ? ? ? ? ? ? format='%(asctime)s - %(levelname)s: %(message)s')
INDEX_URL = 'https://dynamic5.scrape.center/api/book/?limit=18&offset={offset}'
DETAIL_URL = 'https://dynamic5.scrape.center/api/book/{id}'
PAGE_SIZE = 18
PAGE_NUMBER = 100
CONCURRENCY = 5

semaphore = asyncio.Semaphore(CONCURRENCY)
session = None

async def scrape_api(url):
? ?async with semaphore:
? ? ? ?try:
? ? ? ? ? ?logging.info('scraping %s', url)
? ? ? ? ? ?async with session.get(url) as response:
? ? ? ? ? ? ? ?return await response.json()
? ? ? ?except aiohttp.ClientError:
? ? ? ? ? ?logging.error('error occurred while scraping %s', url, exc_info=True)

async def scrape_index(page):
? ?url = INDEX_URL.format(offset=PAGE_SIZE * (page - 1))
? ?return await scrape_api(url)

async def main():
? ?global session
? ?session = aiohttp.ClientSession()
? ?scrape_index_tasks = [asyncio.ensure_future(scrape_index(page)) for page in range(1, PAGE_NUMBER + 1)]
? ?results = await asyncio.gather(*scrape_index_tasks)
? ?logging.info('results %s', json.dumps(results, ensure_ascii=False, indent=2))
? ?

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

到此這篇關(guān)于Python中aiohttp的簡單使用的文章就介紹到這了,更多相關(guān)Python aiohttp 內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: