Python aiohttp
最后修改于 2024 年 1 月 29 日
在本文中，我们将展示如何使用 aiohttp 模块在 Python 中创建异步 HTTP 客户端和服务器。
通过异步编程，我们可以与主程序执行并发地执行任务。
使用 aiohttp 模块，我们可以在 Python 中创建异步 HTTP 客户端和服务器。 该模块还支持 websocket。 它允许创建带有可插拔中间件和路由的 Web 服务器。
Python aiohttp 简单客户端
第一个例子是一个简单的异步 HTTP 客户端。
#!/usr/bin/python
import aiohttp
import asyncio
url = 'http://webcode.me'
async def main():
async with aiohttp.ClientSession() as session:
async with session.get(url) as response:
print("Status:", response.status)
data = await response.text()
print(data)
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
loop.run_until_complete(main())
在该程序中，我们连接到一个小型网站并检索其状态和内容。
import aiohttp import asyncio
我们导入 aiohttp 和 asyncio 模块。
async with aiohttp.ClientSession() as session:
创建一个客户端会话。 它使用连接池进行网络连接。
async with session.get(url) as response:
我们使用 get 方法创建一个 GET 请求。
print("Status:", response.status)
从响应对象，我们通过 status 字段获取状态码。
data = await response.text() print(data)
我们使用 text 方法获取内容并将其打印到终端。
loop = asyncio.new_event_loop() asyncio.set_event_loop(loop) loop.run_until_complete(main())
在 asyncio 模块的帮助下，我们建立一个事件循环，这对于异步编程是必要的。
λ ./simple_client.py Status: 200 <!DOCTYPE html> <html lang="en"> <head> ...
Python aiohttp 多个异步请求
在下一个示例中，我们使用 aiohttp 生成多个异步请求。
#!/usr/bin/python
import aiohttp
import asyncio
async def get_async(url):
async with aiohttp.ClientSession() as session:
return await session.get(url)
urls = ['http://webcode.me', 'https://httpbin.org/get',
'https://google.com', 'https://stackoverflow.com',
'https://github.com']
async def launch():
resps = await asyncio.gather(*map(get_async, urls))
data = [resp.status for resp in resps]
for status_code in data:
print(status_code)
asyncio.run(launch())
在该程序中，我们创建到多个 URL 的异步 HTTP GET 请求。
urls = ['http://webcode.me', 'https://httpbin.org/get',
'https://google.com', 'https://stackoverflow.com',
'https://github.com']
这是 URL 列表。
async def launch():
resps = await asyncio.gather(*map(get_async, urls))
data = [resp.status for resp in resps]
for status_code in data:
print(status_code)
使用内置的 map 函数，我们将 get_async 函数应用于 URL 列表。 返回的列表通过 *（星号）运算符解包为位置参数。 如果所有协程都成功完成，则结果是返回值的聚合列表（HTML 代码）。
asyncio.run(launch())
asyncio.run 是一个方便的函数，可以简化我们的代码。 该函数创建一个事件循环，安排协程，最后关闭循环。
$ ./multiple.py 200 200 200 200 200
Python aiohttp 简单 Web 服务器
以下示例创建一个简单的 Web 服务器。
#!/usr/bin/python
from aiohttp import web
async def home(req):
return web.Response(text="home page")
app = web.Application()
app.add_routes([web.get('/', home)])
web.run_app(app)
Web 服务器具有一个返回文本消息的路由。
async def home(req):
return web.Response(text="home page")
home 函数返回一个文本响应。
app = web.Application()
创建一个 Web 应用程序。
app.add_routes([web.get('/', home)])
使用 add_routes 添加一个新路由。 / 路径映射到 home 处理程序。
web.run_app(app)
run_app 启动 Web 应用程序。
$ ./simple_web.py ======== Running on http://0.0.0.0:8080 ======== (Press CTRL+C to quit)
我们启动 Web 服务器。
$ curl localhost:8080 home page
在不同的终端中，我们使用 curl 工具生成一个 GET 请求。
我们可以使用 Python 装饰器来定义路由。
#!/usr/bin/python
from aiohttp import web
routes = web.RouteTableDef()
@routes.get('/')
async def home(request):
return web.Response(text="home page")
app = web.Application()
app.add_routes(routes)
web.run_app(app)
该程序创建一个简单的 Web 服务器，该服务器具有一个路由。 该路由使用 @routes 装饰器定义。
Python aiohttp JSON 响应
JSON（JavaScript 对象表示法）是一种轻量级的数据交换格式。 JSON 的官方 Internet 媒体类型是 application/json。
aiohttp 模块使用 web.json_response 生成 JSON 响应。
#!/usr/bin/python
from aiohttp import web
routes = web.RouteTableDef()
@routes.get('/')
async def home(req):
return web.Response(text="home page")
@routes.get('/users')
async def users(req):
users = [{'name': 'John Doe', 'email': 'john.doe@example.org'},
{'name': 'Roger Roe', 'email': 'roger.roe@example.org'}]
return web.json_response(users)
app = web.Application()
app.add_routes(routes)
web.run_app(app)
在应用程序中，我们使用装饰器定义两个路由。 /users 路径将 JSON 响应返回给调用者。
$ curl localhost:8080/
home page
$ curl localhost:8080/users
[{"name": "John Doe", "email": "john.doe@example.org"},
{"name": "Roger Roe", "email": "roger.roe@example.org"}]
来源
在本文中，我们使用了 Python 的 aiohttp 模块。
作者
列出所有 Python 教程。