如何快速實現REST API集成以優化業務流程
你可能也喜歡這些API文章!
輸出結果：
運行后，你會看到GitHub上aiohttp倉庫的詳細信息（JSON格式）。是不是很酷？
接下來，我們用aiohttp搭建一個簡易的Web服務。它會根據URL路徑，動態生成響應。
fromaiohttpimportweb
# 請求處理函數
asyncdefhandle(request):
name = request.match_info.get('name',"Stranger")
returnweb.Response(text=f"Hello,{name}!")
# 創建應用
app = web.Application()
app.router.add_get('/', handle)
app.router.add_get('/{name}', handle)
# 啟動服務器
if__name__ =='__main__':
web.run_app(app)訪問說明：
?打開http://localhost:8080/，返回Hello, Stranger!
?打開http://localhost:8080/Alice，返回Hello, Alice!
想象一下，你需要一次性從多個API獲取數據，這時aiohttp的并發處理能力就派上用場了。
importaiohttp
importasyncio
# 異步獲取數據
asyncdeffetch_data(url, session):
asyncwithsession.get(url)asresponse:
returnawaitresponse.text()
# 并發請求多個URL
asyncdeffetch_all(urls):
asyncwithaiohttp.ClientSession()assession:
tasks = [fetch_data(url, session)forurlinurls]
returnawaitasyncio.gather(*tasks)
# 主函數
asyncdefmain():
urls = [
"https://jsonplaceholder.typicode.com/posts/1",
"https://jsonplaceholder.typicode.com/posts/2",
"https://jsonplaceholder.typicode.com/posts/3",
]
results =awaitfetch_all(urls)
forresultinresults:
print(result)
# 運行異步任務
asyncio.run(main())輸出：
程序會并發請求三個API，打印返回的JSON數據，大大提升效率！
中間件可以幫助我們在處理請求前后插入自定義邏輯，比如記錄請求耗時。
fromaiohttpimportweb
importtime
# 自定義中間件
@web.middleware
asyncdeftiming_middleware(request, handler):
start = time.time()
response =awaithandler(request)
duration = time.time() - start
response.headers['X-Process-Time'] = str(duration)
returnresponse
# 創建應用并添加中間件
app = web.Application(middlewares=[timing_middleware])
# 添加路由
asyncdefhandle(request):
returnweb.Response(text="Hello, aiohttp!")
app.router.add_get('/', handle)
# 啟動服務器
if__name__ =='__main__':
web.run_app(app)效果：
每個響應都會包含一個X-Process-Time頭，記錄請求處理時間。
最后，我們用aiohttp實現一個簡單的RESTful API服務。它允許我們創建、讀取和管理“任務”。
fromaiohttpimportweb
importjson
# 數據存儲
tasks = {}
# 獲取任務
asyncdefget_task(request):
task_id = request.match_info['id']
iftask_idintasks:
returnweb.Response(
text=json.dumps(tasks[task_id]),
content_type='application/json'
)
returnweb.Response(status=404)
# 創建任務
asyncdefcreate_task(request):
data =awaitrequest.json()
task_id = str(len(tasks) +1)
tasks[task_id] = data
returnweb.Response(
text=json.dumps({'id': task_id}),
content_type='application/json'
)
# 初始化應用和路由
app = web.Application()
app.router.add_get('/tasks/{id}', get_task)
app.router.add_post('/tasks', create_task)
# 啟動服務器
if__name__ =='__main__':
web.run_app(app)使用方法：
1.創建任務：發送POST請求到/tasks，請求體為JSON格式，如{“name”: “Buy groceries”}。
2.獲取任務：訪問GET /tasks/{id}，返回任務詳情。
示例：
1.創建任務：
?請求體：{“name”: “Do laundry”}
?響應：{“id”: “1”}
2.獲取任務：
?URL：/tasks/1
?響應：{“name”: “Do laundry”}
aiohttp不僅是Python異步編程的王牌工具，更是開發現代Web應用的絕佳選擇。它的優點包括：
?高性能：處理大量并發請求毫無壓力。
?易用性：簡單直觀的API設計，新手也能快速上手。
?擴展性：支持中間件、自定義路由，適配各種應用場景。
文章轉自微信公眾號@柳如不是
