官方文档：Client Quickstart — aiohttp 3.14.0 documentation

aiohttp 客户端快速入门

迫不及待想要上手？本页将为你介绍如何快速入门 aiohttp 客户端 API。

首先，请确保已安装 aiohttp 且版本为最新。

让我们从几个简单的例子开始吧。

发送请求

首先，导入 aiohttp 模块和 asyncio：

import aiohttp
import asyncio

接下来，我们尝试获取一个网页。例如，查询 http://httpbin.org/get：

async def main():
    async with aiohttp.ClientSession() as session:
        async with session.get('http://httpbin.org/get') as resp:
            print(resp.status)
            print(await resp.text())

asyncio.run(main())

现在，我们创建了一个名为 session 的 ClientSession（客户端会话）和一个名为 resp 的 ClientResponse（客户端响应）对象。我们可以从响应中获取所有需要的信息。ClientSession.get() 协程的必需参数是一个 HTTP URL（可以是 str 或 yarl.URL 类实例）。

要发送 HTTP POST 请求，请使用 ClientSession.post() 协程：

session.post('http://httpbin.org/post', data=b'data')

其他 HTTP 方法同样可用：

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')

为了简化对同一站点的多次请求，可以使用 ClientSession 构造函数的 base_url 参数。例如，要请求 http://httpbin.org 的不同端点（endpoint），可以使用以下代码：

async with aiohttp.ClientSession('http://httpbin.org') as session:
    async with session.get('/get'):
        pass
    async with session.post('/post', data=b'data'):
        pass
    async with session.put('/put', data=b'data'):
        pass

注意
不要为每个请求创建一个会话。大多数情况下，你需要的是一个应用对应一个会话，由该会话执行所有请求。
更复杂的场景可能需要按站点创建会话，例如一个用于 Github，一个用于 Facebook。总之，为每个请求创建一个会话是非常糟糕的做法。
一个会话内部包含一个连接池。连接复用（connection reusage）和 keep-alive（默认均开启）可以显著提升整体性能。

会话的上下文管理器用法不是强制的，但这种情况下必须调用 await session.close() 方法，例如：

session = aiohttp.ClientSession()
async with session.get('...'):
    # ...
await session.close()

在 URL 中传递参数

很多时候，你希望在 URL 的查询字符串（query string）中发送一些数据。如果手动构造 URL，这些数据会以键值对的形式出现在问号之后，例如 httpbin.org/get?key=val。aiohttp 允许你通过 params 关键字参数以字典形式提供这些参数。例如，如果要将 key1=value1 和 key2=value2 传递给 httpbin.org/get，可以使用以下代码：

params = {'key1': 'value1', 'key2': 'value2'}
async with session.get('http://httpbin.org/get',
    params=params) as resp:
    expect = 'http://httpbin.org/get?key1=value1&key2=value2'
    assert str(resp.url) == expect

通过打印 URL 可以看到，URL 已经被正确编码了。

对于同一个键包含多个值的情况，可以使用 MultiDict（多值字典）；同时支持嵌套列表的写法（{'key': ['value1', 'value2']}）。

也可以传入一个由 2 元组组成的列表作为参数，这样可以为每个键指定多个值：

params = [('key', 'value1'), ('key', 'value2')]
async with session.get('http://httpbin.org/get',
    params=params) as r:
    expect = 'http://httpbin.org/get?key=value2&key=value1'
    assert str(r.url) == expect

你还可以将字符串内容作为参数传入。值会被用作查询字符串，但传入 params 不会禁用 URL 规范化（canonicalization）。注意，+ 不会被编码：

async with session.get('http://httpbin.org/get',
    params='key=value+1') as r:
    assert str(r.url) == 'http://httpbin.org/get?key=value+1'

注意
aiohttp 在发送请求之前会在内部执行 URL 规范化。
规范化（Canonicalization）使用 IDNA 编解码器对主机部分进行编码，并对路径（path）和查询（query）部分进行重新转义。
例如，URL('http://example.com/путь/%30?a=%31') 会被转换为 URL('http://example.com/%D0%BF%D1%83%D1%82%D1%8C/0?a=1')。
有时候规范化并不是我们想要的，如果服务器可以接受精确的表示并且不会对 URL 进行重新转义。
要禁用规范化，请使用 URL 构造时的 encoded=True 参数：
await session.get(
    URL('http://example.com/%30', encoded=True))

警告
传入 params 会覆盖 encoded=True。如果你需要保留精确的查询字符串字节，绝对不要同时使用这两个选项。请直接构造完整的 URL（包括查询部分）。

响应内容和状态码

我们可以读取服务器响应的内容和状态码。再以 GitHub 为例：

async with session.get('https://api.github.com/events') as resp:
    print(resp.status)
    print(await resp.text())

输出大致如下：

200
'[{"created_at":"2015-06-12T14:06:22Z","public":true,"actor":{...

aiohttp 会自动对服务器返回的内容进行解码。你可以为 text() 方法指定自定义编码：

await resp.text(encoding='windows-1251')

二进制响应内容

你也可以将响应体（response body）作为字节获取，用于非文本请求：

print(await resp.read())
b'[{"created_at":"2015-06-12T14:06:22Z","public":true,"actor":{...

gzip 和 deflate 传输编码（transfer-encoding）会被自动解码。

如果你需要支持 brotli 传输编码，只需安装 Brotli 或 brotlicffi 即可。

如果你需要支持 zstd 传输编码，只需安装 backports.zstd。如果你在使用 Python >= 3.14，则无需任何依赖。

JSON 请求

会话的任意请求方法，如 request()、ClientSession.get()、ClientSession.post() 等，都接受 json 参数：

async with aiohttp.ClientSession() as session:
    async with session.post(url, json={'test': 'object'})

默认情况下，会话使用 Python 标准库的 json 模块进行序列化。但也可以使用其他序列化工具。ClientSession 接受 jsonserialize 和 jsonserialize_bytes 参数：

import orjson

async with aiohttp.ClientSession(
    json_serialize_bytes=orjson.dumps) as session:
    await session.post(url, json={'test': 'object'})

注意
orjson 库比标准 json 更快，且维护活跃。由于 orjson.dumps 返回的是字节，因此应通过 json_serialize_bytes 参数传入，以避免不必要的编码/解码开销。

JSON 响应内容

如果你正在处理 JSON 数据，aiohttp 还内置了 JSON 解码器：

async with session.get('https://api.github.com/events') as resp:
    print(await resp.json())

如果 JSON 解码失败，json() 会抛出一个异常。你可以为 json() 调用指定自定义编码和解码函数。

注意
上述方法会将整个响应体读取到内存中。如果你打算读取大量数据，请考虑使用下文文档中介绍的流式响应（streaming response）方法。

流式响应内容

虽然 read()、json() 和 text() 方法非常方便，但你应该谨慎使用。这些方法都会将整个响应加载到内存中。例如，如果你想下载几个 GB 大小的文件，这些方法会把所有数据加载到内存中。你可以使用 content 属性来代替。它是 aiohttp.StreamReader 类的实例。gzip 和 deflate 传输编码内容会被自动解码：

async with session.get('https://api.github.com/events') as resp:
    await resp.content.read(10)

然而，通常你应该使用以下模式来保存流式数据到文件：

with open(filename, 'wb') as fd:
    async for chunk in resp.content.iter_chunked(chunk_size):
        fd.write(chunk)

一旦显式地从 content 读取数据后，就不能再使用 read()、json() 和 text() 了。

更复杂的 POST 请求

通常，你想要发送一些表单编码的数据——类似于 HTML 表单。为此，只需将字典传递给 data 参数即可。你的数据字典会在请求发送时被自动编码：

payload = {'key1': 'value1', 'key2': 'value2'}
async with session.post('http://httpbin.org/post',
    data=payload) as resp:
    print(await resp.text())

{
    ...
    "form": {
        "key2": "value2",
        "key1": "value1"
    },
    ...
}

如果你想发送非表单编码的数据，可以传入 bytes 而不是字典。这种数据会直接被发送，Content-Type 默认会被设置为 application/octet-stream：

async with session.post(url, data=b'\x00Binary-data\x00') as resp:
    ...

如果你想发送 JSON 数据：

async with session.post(url, json={'example': 'test'}) as resp:
    ...

如果你想发送文本并设置适当的 Content-Type，只需使用 data 参数：

async with session.post(url, data='Тест') as resp:
    ...

POST 上传多部分编码文件

要上传多部分编码（Multipart-encoded）文件：

url = 'http://httpbin.org/post'
files = {'file': open('report.xls', 'rb')}

await session.post(url, data=files)

你也可以显式地设置 filename 和 content_type：

url = 'http://httpbin.org/post'
data = aiohttp.FormData()
data.add_field('file',
    open('report.xls', 'rb'),
    filename='report.xls',
    content_type='application/vnd.ms-excel')

await session.post(url, data=data)

如果你将文件对象作为 data 参数传入，aiohttp 会自动将其流式上传到服务器。支持的格式信息请参阅 StreamReader。

流式上传

aiohttp 支持多种类型的流式上传（streaming uploads），可以让你在发送大文件时无需将文件全部读入内存。

作为简单的例子，只需为请求体提供一个类文件对象即可：

with open("massive-body", "rb") as f:
    await session.post("https://httpbin.org/post", data=f)

你也可以提供一个异步生成器（asynchronous generator），例如实时生成数据：

async def data_generator():
    for i in range(10):
        yield f"line {i}\n".encode()

async with session.post("https://httpbin.org/post",
    data=data_generator()) as resp:
    print(await resp.text())

警告
异步生成器和其他不可回绕（non-rewindable）的数据源（如 StreamReader）在发生重定向（例如 HTTP 307 或 308）时无法重放。如果请求体已经被流式发送，aiohttp 会抛出 ClientPayloadError。
如果你的端点可能会发生重定向，请选择以下方案之一：
传入支持寻址（seekable）的类文件对象或 bytes。
使用 allow_redirects=False 禁用自动重定向，并手动处理重定向。

由于 content 属性是一个 StreamReader（提供异步迭代器协议），你可以将 GET 和 POST 请求链接在一起：

resp = await session.get('http://python.org')
await session.post('http://httpbin.org/post',
    data=resp.content)

注意
Python 3.5 原生不支持异步生成器，可以使用 async_generator 库作为变通方案。

自 3.1 版本起弃用：aiohttp 仍然支持 aiohttp.streamer 装饰器，但这种方法已被弃用，推荐使用上述示例所示的异步生成器。

WebSocket

aiohttp 原生支持客户端 WebSocket。

你需要使用 aiohttp.ClientSession.ws_connect() 协程来建立客户端 WebSocket 连接。它接受一个 URL 作为第一个参数，并返回一个 ClientWebSocketResponse 对象。使用该对象的方法，你可以与 WebSocket 服务器进行通信：

async with session.ws_connect('http://example.org/ws') as ws:
    async for msg in ws:
        if msg.type == aiohttp.WSMsgType.TEXT:
            if msg.data == 'close cmd':
                await ws.close()
                break
            else:
                await ws.send_str(msg.data + '/answer')
        elif msg.type == aiohttp.WSMsgType.ERROR:
            break

你必须只用一个 WebSocket 任务来进行读取（例如 await ws.receive() 或 async for msg in ws:），但可以拥有多个（只能异步发送数据的）写入任务（例如 await ws.send_str('data')）。

超时

超时设置存储在 ClientTimeout（客户端超时）数据结构中。

默认情况下，aiohttp 使用 300 秒（5 分钟）的总超时时间，这意味着整个操作应在 5 分钟内完成。为了给 DNS 备用方案留出时间，默认的 sock_connect 超时为 30 秒。

可以通过会话的 timeout 参数覆盖该值（单位为秒）：

timeout = aiohttp.ClientTimeout(total=60)
async with aiohttp.ClientSession(timeout=timeout) as session:
    ...

超时也可以在单个请求中被覆盖，例如 ClientSession.get()：

async with session.get(url, timeout=timeout) as resp:
    ...

ClientTimeout 支持的字段如下：

total
整个操作的最大秒数，包括建立连接、发送请求和读取响应。
connect
建立新连接或等待连接池中有空闲连接（如果达到连接池限制）的最大秒数。
sock_connect
建立新连接（而非从连接池获取）时与对等端连接的最大秒数。
sock_read
从对等端读取新数据片段之间允许的最大秒数。
ceil_threshold
触发绝对超时值（absolute timeout values）向上取整的阈值。

所有字段都是浮点数，None 或 0 表示禁用该特定超时检查，有关默认值和更多详情请参阅 ClientTimeout 参考文档。

因此，默认超时值为：

aiohttp.ClientTimeout(total=5*60, connect=None,
    sock_connect=None, sock_read=None, ceil_threshold=5)

注意
如果超时值大于或等于 5 秒，aiohttp 会对超时进行向上取整。超时会发生在 current_time + timeout 之后下一个整数秒的时刻。
这样做是为了优化：当许多并发任务被调度在几乎相同但略有不同的绝对时间唤醒时，会导致大量的事件循环唤醒，严重影响性能。
该优化通过将绝对唤醒时间调整到与其他任务完全相同的时刻来实现，事件循环每秒最多唤醒一次来检查超时是否到期。
较小的超时值不会被取整，以利于测试；而在实际网络中，超时通常大于几十秒。不过，默认的 5 秒阈值可以通过 ceil_threshold 参数进行配置。

2026年6月

aiohttp 入门指南