第 6 章：异步 HTTP 请求与客户端复用

学生信息查询助手现在已经能做到三件事：

1. 批量查询学生信息。
2. 使用 Semaphore 控制并发数量。
3. 使用超时和异常处理，避免一个慢查询拖住整个程序。

但是前面的“外部查询”仍然只是用 asyncio.sleep() 模拟的。

真实项目里，学生信息通常不是写死在本地代码里，而是来自外部 HTTP 接口：

• 学生基础信息来自学生服务。
• 学生课程信息来自课程服务。
• 学生积分信息来自积分服务。

如果我们在 async def 里继续写同步请求，比如：

requests.get(url)

它会阻塞当前线程，也会阻塞事件循环。这样虽然函数写成了 async def，但异步优势会被同步阻塞代码直接破坏。

本章要解决 4 个问题：

1. 异步项目里为什么不要直接用 requests.get()？
2. 如何安装并使用 httpx.AsyncClient？
3. 如何复用 HTTP 客户端，而不是每次请求都新建客户端？
4. 如何把真实 HTTP 请求纳入已有的并发限制、超时和统一结果结构？

本章只解决异步 HTTP 请求的日常项目写法。

本章不讲 FastAPI，不讲数据库，不讲 Redis，不讲 HTTP 协议细节，也不讲 httpx 的完整参数表。

本章学习目标

学完本章后，我们要能做到：

1. 理解同步 HTTP 请求为什么会阻塞异步程序。
2. 知道 httpx.AsyncClient 是常见的异步 HTTP 客户端。
3. 能安装 httpx，并写出最小异步 GET 请求。
4. 能使用 async with httpx.AsyncClient(...) as client: 管理客户端。
5. 能使用 await client.get(url) 发起异步请求。
6. 能使用 response.status_code 判断响应状态。
7. 能使用 response.json() 读取 JSON 响应。
8. 能复用同一个 AsyncClient 请求多个接口。
9. 能把 AsyncClient 与 gather()、Semaphore、超时和异常处理组合到小案例中。
10. 能排查忘记 await、每次请求都新建客户端、继续使用 requests 等常见错误。

HTTP 学生信息聚合场景

现在我们要把“学生信息查询助手”升级一步。

前面查询学生信息时，我们写的是：

await asyncio.sleep(0.2)
return student_data.get(student_id)

这只是模拟等待。

真实项目更像下面这样：

查询学生汇总信息
    ├── 请求学生基础信息接口
    ├── 请求学生课程信息接口
    └── 请求学生积分信息接口

可以画成一个清晰的扇出聚合管道：

这三个接口都属于 I/O 等待场景。

所以它们适合用异步 HTTP 客户端请求。

本章我们使用 httpx.AsyncClient。它的写法和前面学过的异步知识可以自然接上：

async with httpx.AsyncClient() as client:
    response = await client.get(url)

注意：这里的重点不是 HTTP 协议，而是“在异步程序中正确请求外部接口”。

异步 HTTP 请求与聚合

下面每一步的完整代码都直接显示在页面里。将代码按标注文件名保存到本地 _code_async/c06_httpx_asyncclient/，然后在 _code_async 目录执行命令。

1. 最小异步 GET 请求

同步请求与异步请求

同步 HTTP 请求可以简单理解为：

发出请求后，当前程序停在这里，等响应回来才继续。

如果在异步函数里写同步请求，就可能阻塞事件循环。

异步 HTTP 请求可以简单理解为：

发出请求后，当前任务等待响应；等待期间，其他异步任务可以继续推进。

本章我们不深入讲底层网络模型，只需要记住一句话：

在 async def 里请求外部 HTTP 接口，优先使用异步 HTTP 客户端。

同步请求的阻塞影响

requests 是同步 HTTP 客户端。

下面这种写法不推荐：

async def fetch_data():
    response = requests.get("https://example.com")
    return response.json()

问题不在于它一定不能跑，而是它会阻塞当前线程。

如果这个函数运行在异步程序中，阻塞会影响其他异步任务。

项目里常见后果：

1. 明明写了 async def，并发效果还是很差。
2. 一个外部接口慢，其他请求也被拖住。
3. Web 接口吞吐下降。
4. 排查时很难看出是同步库阻塞了事件循环。

本章先把规则记住：

异步函数里请求 HTTP，优先用 httpx.AsyncClient，不要直接塞 requests.get()。

客户端创建与响应读取

httpx.AsyncClient 是 httpx 提供的异步 HTTP 客户端。

它解决的问题是：

1. 在异步函数里发 HTTP 请求。
2. 用 await 等待响应。
3. 复用客户端连接。
4. 设置请求超时。
5. 配合 gather()、Semaphore 做并发请求。

最小结构是：

import httpx

async with httpx.AsyncClient(timeout=3.0) as client:
    response = await client.get(url)

这里有三个关键点：

1. AsyncClient 是异步客户端。
2. await client.get(url) 才是真正等待 HTTP 响应。
3. async with 负责管理客户端生命周期，使用完会关闭。

需求：先写一个最小异步 HTTP GET 请求，确认 httpx.AsyncClient 的基本形态。

功能分析：

1. 安装 httpx。
2. 使用 httpx.AsyncClient 创建异步 HTTP 客户端。
3. 使用 await client.get(url) 发起请求。
4. 打印响应状态码和 JSON 内容。
5. 使用 asyncio.run(main()) 启动异步程序。

文件名：01_async_http.py

安装命令：

uv sync

代码：

01_async_http.py

import asyncio

import httpx


async def main():
    url = "https://httpbin.org/json"

    async with httpx.AsyncClient(timeout=3.0) as client:
        response = await client.get(url)
        print("状态码：", response.status_code)
        print("JSON 数据：", response.json())


if __name__ == "__main__":
    asyncio.run(main())

运行命令：

uv run python c06_httpx_asyncclient/01_async_http.py

运行结果说明：

如果网络正常，可以看到类似结果：

状态码： 200
JSON 数据： {'slideshow': {...}}

如果网络不稳定，可能出现连接失败或超时。这不一定是代码写错，也可能是访问公网测试接口失败。

这个最小示例请求的是公网 https://httpbin.org/json，运行前需要能访问该地址。验收时重点看是否得到 HTTP 状态码和包含 slideshow 的 JSON；公网不可达时，可继续使用后面的本地学生服务示例验证异步请求写法。

若提示 Using SOCKS proxy, but the 'socksio' package is not installed，说明环境代理与可选依赖不匹配，应检查代理配置；这与查询函数的异步语法无关。后面的本地示例使用 trust_env=False，不读取环境代理。

这段代码要看懂 4 个点：

1. httpx.AsyncClient 是异步 HTTP 客户端。
2. await client.get(url) 表示等待 HTTP 响应。
3. response.status_code 是 HTTP 状态码。
4. response.json() 把 JSON 响应转成 Python 数据。

注意：真实项目里，不建议每次请求都临时创建一个新的客户端。本节只是最小演示，下一节会改成复用客户端。

2. 本地学生服务的准备

服务文件与接口约定

从这一章开始，我们直接请求一个单独启动的学生服务。

这里把 FastAPI 服务作为已经提供的测试工具，不展开服务端框架语法。保存下面的文件，按命令启动并验证接口即可；异步 Web 接口的编写留到第 7 章。

你只需要知道两件事：

1. 课程会提供 _code_async/c06_httpx_asyncclient/02_external_student_server/main.py。
2. 同目录的 api_spec.md 说明客户端要请求哪些接口。

本章重点仍然是：

如何使用 httpx.AsyncClient 发起异步 HTTP 请求，并把结果整理成项目可用的结构。

需求：准备一个独立运行的本地学生服务，先验证接口，再用下一节的客户端查询三类信息。

功能分析：

1. 提供一个 02_external_student_server/，里面用 main.py 启动服务。
2. 提供 3 个学生接口：profile、courses、points。
3. 对不存在的学生返回 404。
4. 普通接口等待 0.2 秒，用 student_id = 3 的积分接口模拟 2 秒慢响应。
5. 同目录再补一份 api_spec.md，让客户端对接更直接。

文件名：_code_async/c06_httpx_asyncclient/02_external_student_server/main.py

代码：

main.py

import asyncio

from fastapi import FastAPI, HTTPException

app = FastAPI()

student_profiles = {
    1: {"id": 1, "name": "张三", "age": 18},
    2: {"id": 2, "name": "李四", "age": 19},
    3: {"id": 3, "name": "王五", "age": 20},
}

student_courses = {
    1: ["Python", "Web 后端"],
    2: ["数据库", "Redis"],
    3: ["异步编程", "接口聚合"],
}

student_points = {
    1: 95,
    2: 88,
    3: 91,
}


def ensure_student_exists(student_id: int):
    if student_id not in student_profiles:
        raise HTTPException(status_code=404, detail="student not found")


@app.get("/students/{student_id}/profile")
async def get_student_profile(student_id: int):
    ensure_student_exists(student_id)
    await asyncio.sleep(0.2)
    return student_profiles[student_id]


@app.get("/students/{student_id}/courses")
async def get_student_courses(student_id: int):
    ensure_student_exists(student_id)
    await asyncio.sleep(0.2)
    return {"courses": student_courses.get(student_id, [])}


@app.get("/students/{student_id}/points")
async def get_student_points(student_id: int):
    ensure_student_exists(student_id)

    if student_id == 3:
        await asyncio.sleep(2)
    else:
        await asyncio.sleep(0.2)

    return {"points": student_points.get(student_id, 0)}


if __name__ == "__main__":
    import uvicorn
    uvicorn.run("main:app", port=8000, reload=True)

服务启动与接口验收

在终端 A 从 _code_async 目录启动这个服务，保持终端运行，后面的客户端示例才能访问它：

uv run python -m uvicorn main:app --port 8000 --app-dir c06_httpx_asyncclient/02_external_student_server

看到服务启动信息后，在另一个终端验证接口：

curl http://127.0.0.1:8000/students/1/profile

应返回 {"id":1,"name":"张三","age":18}。若连接失败，先检查终端 A 中的服务是否成功启动；通过后再运行客户端。服务持续占用终端 A，客户端在终端 B 运行。

接口说明：_code_async/c06_httpx_asyncclient/02_external_student_server/api_spec.md

完整接口说明文件：

api_spec.md

# 第六章外部学生服务接口说明

默认你已经会自行启动 FastAPI 服务。

这个服务只负责给第六章客户端示例提供 3 个接口。

## 基础信息

- 默认地址：`http://127.0.0.1:8000`
- 成功状态码：`200`
- 学生不存在：`404`

## 接口列表

### `GET /students/{student_id}/profile`

成功响应：

```json
{
  "id": 1,
  "name": "张三",
  "age": 18
}
```

### `GET /students/{student_id}/courses`

成功响应：

```json
{
  "courses": ["Python", "Web 后端"]
}
```

### `GET /students/{student_id}/points`

成功响应：

```json
{
  "points": 95
}
```

## 特殊约定

- 存在学生的普通查询人为等待 0.2 秒，用来比较三个请求顺序执行与并发执行的耗时。
- `student_id = 3` 时，`/points` 接口等待 2 秒，用来演示客户端超时处理。
- 第六章完整案例默认用 `http://127.0.0.1:8000` 作为 `base_url`。

# 第六章外部学生服务接口说明

默认地址：http://127.0.0.1:8000

GET /students/{student_id}/profile
返回学生基础信息对象

GET /students/{student_id}/courses
返回 {"courses": [...]}

GET /students/{student_id}/points
返回 {"points": 95}

特殊约定：
student_id = 3 时，points 接口会故意慢响应
普通接口人为等待 0.2 秒
学生不存在时返回 404

注意：

1. 本节只要求按提供的命令启动测试服务并验证接口。
2. 这里不展开讲服务端框架语法。
3. 本章重点仍然是客户端如何异步请求这些接口。

3. 多接口请求中的客户端复用

客户端生命周期与连接复用

不要每个请求都这样写：

async with httpx.AsyncClient() as client:
    ...

如果每个小函数内部都新建客户端，会增加连接创建和关闭的成本。

更推荐的写法是：

async def fetch_student_profile(client, student_id):
    response = await client.get(f"/students/{student_id}/profile")
    return response.json()

然后在外层只创建一次客户端：

async with httpx.AsyncClient(...) as client:
    profile = await fetch_student_profile(client, 1)

简单理解：

客户端像一个可复用的“请求工具”。一组相关请求最好共用同一个客户端。

本章只讲够用原则，不展开连接池内部实现。

HTTPX 超时与代码块超时

httpx.AsyncClient(timeout=3.0) 表示 HTTP 客户端层面的超时配置。

这个值分别用于连接、读取、写入和连接池等待。其中读取超时限制的是等待一块响应数据的时间，因此它不表示“整次请求的总耗时最多 3 秒”。HTTPX 超时会抛出 httpx.TimeoutException 的子类，本章统一捕获 httpx.TimeoutException。

上一章的 asyncio.timeout() 则为整个异步代码块设置期限，退出超时块时向外抛出内置的 TimeoutError。如果需要限制一组请求的总等待时间，可以在客户端超时之外再包一层 asyncio.timeout()；本章完整案例只配置 HTTPX 客户端超时。

这些 HTTPX 超时也不包含等待 Semaphore 名额的时间。具体规则见 HTTPX 超时文档，本章不展开逐项配置。

需求：一个学生汇总信息要从 3 个接口获取：基础信息、课程信息、积分信息。我们希望复用同一个 HTTP 客户端，并同时请求这 3 个接口。

功能分析：

1. 写 3 个请求函数。
2. 每个函数都接收同一个 client。
3. 外层只创建一次 httpx.AsyncClient。
4. 使用 asyncio.gather() 同时等待多个 HTTP 请求。
5. 本节主要看结构，完整稳定案例会直接请求本地学生服务。

文件名：03_reuse_client.py

安装命令：

uv sync

代码：

03_reuse_client.py

import asyncio
import time

import httpx

BASE_URL = "http://127.0.0.1:8000"


async def fetch_student_profile(client, student_id):
    response = await client.get(f"{BASE_URL}/students/{student_id}/profile")
    return response.json()


async def fetch_student_courses(client, student_id):
    response = await client.get(f"{BASE_URL}/students/{student_id}/courses")
    return response.json()


async def fetch_student_points(client, student_id):
    response = await client.get(f"{BASE_URL}/students/{student_id}/points")
    return response.json()


async def main():
    student_id = 1

    async with httpx.AsyncClient(timeout=3.0, trust_env=False) as client:
        start_time = time.perf_counter()
        profile, courses, points = await asyncio.gather(
            fetch_student_profile(client, student_id),
            fetch_student_courses(client, student_id),
            fetch_student_points(client, student_id),
        )
        end_time = time.perf_counter()

    print("基础信息接口结果：", profile)
    print("课程信息接口结果：", courses)
    print("积分信息接口结果：", points)
    print(f"请求总耗时：{end_time - start_time:.2f} 秒")


if __name__ == "__main__":
    asyncio.run(main())

保持上一节的测试服务在终端 A 运行。接口验收通过后，在终端 B 从本地 _code_async 目录运行客户端：

uv run python c06_httpx_asyncclient/03_reuse_client.py

运行结果说明：

服务启动后，终端 B 应看到 1 号学生的三类结果：

基础信息接口结果： {'id': 1, 'name': '张三', 'age': 18}
课程信息接口结果： {'courses': ['Python', 'Web 后端']}
积分信息接口结果： {'points': 95}
请求总耗时：0.20 秒

本节固定请求存在的学生，重点验证同一个客户端能取得三份 JSON；不存在学生的 404 和超时会在完整案例中处理。

本地演示服务的三个普通接口分别等待 0.2 秒，便于比较顺序请求与并发请求。计时范围包含三个请求及响应处理，不包含客户端创建与关闭；正常并发查询耗时接近 0.2 秒，具体数值会有网络与调度开销。

这个例子主要看结构：

创建一个 AsyncClient
    ├── 请求 profile
    ├── 请求 courses
    └── 请求 points

注意：

1. client 在 main() 中只创建一次。
2. 3 个请求函数都接收同一个 client。
3. gather() 适合这种“同时请求多个相关接口”的场景。
4. 这里不重复讲 gather() 细节，第 3 章已经讲过。
5. 如果某个接口失败，后面完整案例会做安全封装。

不要在每个函数里都写：

async with httpx.AsyncClient() as client:
    ...

这样会让客户端反复创建和关闭，不适合项目代码。

4. 学生 HTTP 信息聚合

需求：完成一个稳定可运行的脚本，从 3 个外部 HTTP 接口聚合学生信息。

本案例要做到：

1. 请求学生基础信息接口。
2. 请求学生课程信息接口。
3. 请求学生积分信息接口。
4. 一个学生的 3 类信息可以同时请求。
5. 多个学生可以批量查询。
6. 使用 Semaphore 控制同时发出的 HTTP 请求数量。
7. 处理不存在学生、HTTP 错误和 HTTP 超时。
8. 复用同一个 AsyncClient。

项目结构：

_code_async/c06_httpx_asyncclient/
├── 01_async_http.py
├── 02_external_student_server/
│   ├── api_spec.md
│   ├── main.py
├── 03_reuse_client.py
└── 04_http_student_summary/
    ├── api_spec.md
    ├── main.py
    └── student_api_server.py

文件名：main.py

main.py

import asyncio
import time

import httpx

BASE_URL = "http://127.0.0.1:8000"


async def fetch_json(client, url, semaphore):
    async with semaphore:
        response = await client.get(url)

    if response.status_code == 404:
        return None

    response.raise_for_status()
    return response.json()


async def fetch_student_profile(client, student_id, semaphore):
    return await fetch_json(client, f"/students/{student_id}/profile", semaphore)


async def fetch_student_courses(client, student_id, semaphore):
    return await fetch_json(client, f"/students/{student_id}/courses", semaphore)


async def fetch_student_points(client, student_id, semaphore):
    return await fetch_json(client, f"/students/{student_id}/points", semaphore)


async def fetch_student_summary(client, student_id, semaphore):
    try:
        results = await asyncio.gather(
            fetch_student_profile(client, student_id, semaphore),
            fetch_student_courses(client, student_id, semaphore),
            fetch_student_points(client, student_id, semaphore),
            return_exceptions=True,
        )
        for result in results:
            if isinstance(result, BaseException):
                raise result
        profile, courses, points = results

        if profile is None:
            return {
                "student_id": student_id,
                "ok": False,
                "reason": "not_found",
                "data": None,
            }

        missing = []
        if courses is None:
            missing.append("课程")
        if points is None:
            missing.append("积分")
        if missing:
            return {
                "student_id": student_id,
                "ok": False,
                "reason": "incomplete_data",
                "data": None,
                "message": f"{'、'.join(missing)}接口返回 404",
            }

        return {
            "student_id": student_id,
            "ok": True,
            "reason": "success",
            "data": {
                "profile": profile,
                "courses": courses["courses"],
                "points": points["points"],
            },
        }

    except httpx.TimeoutException as exc:
        return {
            "student_id": student_id,
            "ok": False,
            "reason": "timeout",
            "data": None,
            "message": str(exc),
        }

    except httpx.HTTPStatusError as exc:
        return {
            "student_id": student_id,
            "ok": False,
            "reason": "http_status_error",
            "data": None,
            "message": str(exc),
        }

    except httpx.HTTPError as exc:
        return {
            "student_id": student_id,
            "ok": False,
            "reason": "http_error",
            "data": None,
            "message": str(exc),
        }

    except asyncio.CancelledError:
        print(f"学生 {student_id} 的查询任务被取消，继续向外抛出")
        raise


async def fetch_students_safely(student_ids, limit=3):
    if limit < 1:
        raise ValueError("limit 必须大于等于 1")

    semaphore = asyncio.Semaphore(limit)

    async with httpx.AsyncClient(
        base_url=BASE_URL, timeout=1.0, trust_env=False
    ) as client:
        tasks = [
            fetch_student_summary(client, student_id, semaphore)
            for student_id in student_ids
        ]
        results = await asyncio.gather(*tasks, return_exceptions=True)
        for result in results:
            if isinstance(result, BaseException):
                raise result
        return results


def format_result(result):
    student_id = result["student_id"]

    if result["ok"]:
        data = result["data"]
        profile = data["profile"]
        courses = "、".join(data["courses"])
        points = data["points"]
        return (
            f"学生 {student_id} 查询成功："
            f"{profile['name']}，年龄 {profile['age']}，"
            f"课程：{courses}，积分：{points}"
        )

    if result["reason"] == "not_found":
        return f"学生 {student_id} 查询失败：学生不存在"

    if result["reason"] == "timeout":
        return f"学生 {student_id} 查询失败：外部接口超时"

    if result["reason"] == "incomplete_data":
        return f"学生 {student_id} 查询失败：{result['message']}"

    return f"学生 {student_id} 查询失败：{result['reason']}"


async def main():
    student_ids = [1, 2, 3, 404]

    start_time = time.perf_counter()
    results = await fetch_students_safely(student_ids, limit=2)
    end_time = time.perf_counter()

    for result in results:
        print(format_result(result))

    print(f"总耗时：{end_time - start_time:.2f} 秒")


if __name__ == "__main__":
    asyncio.run(main())

文件名：_code_async/c06_httpx_asyncclient/04_http_student_summary/student_api_server.py

student_api_server.py

import asyncio

from fastapi import FastAPI, HTTPException

app = FastAPI()

student_profiles = {
    1: {"id": 1, "name": "张三", "age": 18},
    2: {"id": 2, "name": "李四", "age": 19},
    3: {"id": 3, "name": "王五", "age": 20},
}

student_courses = {
    1: ["Python", "Web 后端"],
    2: ["数据库", "Redis"],
    3: ["异步编程", "接口聚合"],
}

student_points = {
    1: 95,
    2: 88,
    3: 91,
}


def ensure_student_exists(student_id: int):
    if student_id not in student_profiles:
        raise HTTPException(status_code=404, detail="student not found")


@app.get("/students/{student_id}/profile")
async def get_student_profile(student_id: int):
    ensure_student_exists(student_id)
    return student_profiles[student_id]


@app.get("/students/{student_id}/courses")
async def get_student_courses(student_id: int):
    ensure_student_exists(student_id)
    return {"courses": student_courses.get(student_id, [])}


@app.get("/students/{student_id}/points")
async def get_student_points(student_id: int):
    ensure_student_exists(student_id)

    if student_id == 3:
        await asyncio.sleep(2)

    return {"points": student_points.get(student_id, 0)}

配套接口说明文件：

api_spec.md

# 第六章完整案例外部服务接口说明

默认客户端地址：`http://127.0.0.1:8000`

## 接口

### `GET /students/{student_id}/profile`

```json
{
  "id": 1,
  "name": "张三",
  "age": 18
}
```

### `GET /students/{student_id}/courses`

```json
{
  "courses": ["Python", "Web 后端"]
}
```

### `GET /students/{student_id}/points`

```json
{
  "points": 95
}
```

## 失败约定

- 学生不存在：`404`
- `student_id = 3` 的积分接口等待 2 秒，用来演示超时处理；其他接口不额外设置等待

## 客户端结果约定

- 基础信息接口返回 404：`reason="not_found"`。
- 基础信息存在，但课程或积分接口返回 404：`reason="incomplete_data"`，`message` 标明缺失的接口。
- HTTPX 请求超时：`reason="timeout"`。
- 其他 HTTP 错误状态：`reason="http_status_error"`。
- 连接等请求错误：`reason="http_error"`。
- 批量查询的并发值必须大于等于 1，无效值立即抛出 `ValueError`。

在本地 _code_async 目录同步依赖：

uv sync

启动命令需要两个终端，均从本地 _code_async 目录开始。

终端 A，启动案例自带的学生服务：

若前面的测试服务仍在终端 A 运行，先按 Ctrl+C 停止它，再执行下面的命令。两个服务都使用 8000 端口，不能同时绑定该端口。完整案例的服务只对学生 3 的积分接口设置 2 秒等待，其他接口不额外等待 0.2 秒。

uv run python -m uvicorn student_api_server:app --port 8000 --app-dir c06_httpx_asyncclient/04_http_student_summary

终端 B，运行聚合脚本：

uv run python c06_httpx_asyncclient/04_http_student_summary/main.py

运行结果说明：

可以看到类似结果：

学生 1 查询成功：张三，年龄 18，课程：Python、Web 后端，积分：95
学生 2 查询成功：李四，年龄 19，课程：数据库、Redis，积分：88
学生 3 查询失败：外部接口超时
学生 404 查询失败：学生不存在
总耗时：1.00 秒

本例耗时通常接近 1 秒，主要来自学生 3 的积分请求触发客户端读取超时；真实 HTTP 请求本身并不一定耗时 1 秒。排队与调度开销会使结果略有变化。

HTTP 状态与统一失败结果

收到响应并不代表业务请求成功。client.get() 返回响应后，先判断课程约定的 404，再通过 response.raise_for_status() 检查其他错误状态；后者会抛出 httpx.HTTPStatusError。response.json() 是同步方法，本章用它读取已经接收完成的 JSON 响应，不需要再写 await。

情况	统一结果	处理依据
三类接口成功	success	收集基础信息、课程和积分。
基础信息返回 404	not_found	按课程接口约定，学生不存在。
基础信息存在，但课程或积分返回 404	incomplete_data	学生存在，但无法取得完整汇总；message 标明缺失接口。
HTTPX 请求超时	timeout	捕获 httpx.TimeoutException。
其他 HTTP 错误状态，如 503	http_status_error	捕获 raise_for_status() 抛出的 httpx.HTTPStatusError。
连接失败等请求错误	http_error	捕获剩余的 httpx.HTTPError。

三个查询都结束后再检查结果；课程或积分缺失时先返回失败结构，不继续对 None 读取字段。取消信号继续向外传播，不转换为这些业务失败。未知 JSON 结构等接口契约错误不在本章的 HTTP 错误处理范围内。

fetch_students_safely() 还会在创建限制器前拒绝小于 1 的并发值，保留前两章的参数校验。

这个完整案例有几个关键点：

1. 本目录的 student_api_server.py 提供了 3 个外部学生接口。
2. fetch_json() 封装单次 GET 请求。
3. fetch_student_profile()、fetch_student_courses()、fetch_student_points() 分别请求 3 类接口。
4. fetch_student_summary() 使用 gather() 同时请求一个学生的 3 类信息。
5. fetch_json() 在每次 client.get() 前获取 Semaphore 名额，因此限制的是同时发出的 HTTP 请求数量。
6. 三个请求通过 gather(..., return_exceptions=True) 全部结束后再检查错误，避免返回失败时仍有同组请求在运行。
7. 代码中的 BaseException 只用于识别异常结果并立即重新抛出，不会吞掉 CancelledError。
8. format_result() 统一打印成功、超时、查询不到等结果。
9. AsyncClient 只创建一次，并在 async with 结束时自动关闭。

本地服务示例只访问本机，所以客户端设置 trust_env=False，避免环境代理影响 127.0.0.1 请求；第 1 个公网请求示例不使用这项设置。

注意：真实项目中，base_url 会换成真实服务地址，但客户端聚合结构基本一致。

本章小结

本章把前面模拟的 I/O 等待，替换成了真实异步 HTTP 请求写法。

本章最核心的是 5 件事：

1. async def 里不要直接塞同步 requests.get()。
2. 异步 HTTP 请求可以使用 httpx.AsyncClient。
3. await client.get(url) 用来等待 HTTP 响应。
4. 同一个 AsyncClient 应该尽量复用。
5. HTTP 请求也可以继续配合 gather()、Semaphore 和基础异常处理。

现在学生信息查询助手已经能从多个 HTTP 接口聚合学生信息。

下一步要把这套逻辑放进 Web 接口里。

练习

练习 1：保持完整案例的客户端写法不变，只在 fetch_student_points() 中把 /points 改成 /points_missing。确认学生 1、2、3 返回 reason="incomplete_data"，打印“积分接口返回 404”；学生 404 仍返回 not_found。恢复地址后，确认原有成功和超时结果重新出现。

练习 2：使用 03_reuse_client.py 和配套的本地测试服务，固定查询学生 1。先运行并发版本，再把 gather() 改成依次 await 三个查询函数，保持客户端和计时位置不变。检查三份 JSON 是否一致，耗时是否从接近 0.2 秒增加到接近 0.6 秒；随后恢复并发版本。

练习 3：在 03_reuse_client.py 每次创建客户端前打印“创建 HTTP 客户端”，确认当前版本只打印一次。再把客户端上下文移入三个请求函数，各自在 async with httpx.AsyncClient(timeout=3.0, trust_env=False) 内创建并关闭客户端，去掉函数的 client 参数和 main() 中的客户端上下文，并同步修改 main() 的三次调用，使其只传入 student_id。确认创建提示变为三次，三份 JSON 仍一致；最后恢复共用客户端。用创建次数验证复用，不以一次运行的耗时判断连接复用的收益。

练习 4：运行完整案例时，先停止终端 A 的学生服务，再运行客户端，确认连接失败被整理为 http_error；重新启动服务后，确认正常结果恢复。连接失败与收到 HTTP 错误状态是不同情况。