用aiohttp实现RESTful API，快速上手Python异步编程

无论你是想构建高性能的Web服务，还是需要并发处理海量的HTTP请求，aiohttp都能成为你的秘密武器。

准备好了吗？

和柳老师一起，从安装到实战，掌握aiohttp的魅力吧！

一、啥是aiohttp？它有多厉害？

aiohttp是一个基于Python的asyncio异步框架，能够用来：

1.作为客户端：快速发送异步HTTP请求，高效抓取数据。

2.作为服务器：构建高性能的Web服务，轻松处理成百上千的并发请求。

aiohttp到底好在哪？

•高效：基于异步IO，适合处理大量并发任务。

•轻量：简单好用，无需复杂配置。

•灵活：支持中间件、信号等扩展功能。

一句话：aiohttp是现代Web开发者不可错过的宝藏工具！

二、安装aiohttp：一秒钟搞定

直接用pip安装：

pip install aiohttp

如果你喜欢用Anaconda，也可以这样安装：

conda install -c conda-forge aiohttp

就这么简单，搞定！接下来，我们直接开始写代码。

三、用aiohttp搞点简单的

1. 客户端：发送异步HTTP请求

我们用aiohttp做一个小实验，请求GitHub的API，获取某个仓库的信息。

importaiohttp



importasyncio







# 异步获取数据



asyncdeffetch_data(url):



   asyncwithaiohttp.ClientSession()assession:



       asyncwithsession.get(url)asresponse:



           returnawaitresponse.text()







# 主函数



asyncdefmain():



   url ="https://api.github.com/repos/aio-libs/aiohttp"



   data =awaitfetch_data(url)



   print(data)







# 运行异步任务



asyncio.run(main())

输出结果：

运行后，你会看到GitHub上aiohttp仓库的详细信息（JSON格式）。是不是很酷？

2. 服务端：构建一个简单的Web服务

接下来，我们用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!

四、aiohttp的高级玩法

1. 同时请求多个URL

想象一下，你需要一次性从多个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数据，大大提升效率！

2. 中间件：为每个请求加点“料”

中间件可以帮助我们在处理请求前后插入自定义逻辑，比如记录请求耗时。

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头，记录请求处理时间。

五、完整案例：打造一个RESTful API

最后，我们用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)