05 · 异步、异常与文件

🎯 难度：初级⏱ 预计：90 分钟# asyncio# 异常# pathlib

异步：熟悉的 await，不同的启动方式

import asyncio

async def greet():
    await asyncio.sleep(0.1)
    return "hello"

async def main():
    result = await greet()
    print(result)

asyncio.run(main())

重要差异：调用 greet() 只创建协程对象，不会像 JS 的 async 函数那样立即执行到第一个 await。你需要 await、创建 Task，或由事件循环调度它。

在普通 Python 脚本里用 asyncio.run(main()) 启动事件循环；FastAPI 这类框架已经帮你启动了，不要在异步路由里再次调用 asyncio.run()。

并发等待多个请求

import asyncio

async def query(city):
    await asyncio.sleep(0.1)
    return f"{city}查询完成"

async def main():
    results = await asyncio.gather(query("上海"), query("北京"))
    print(results)  # 顺序和输入对应

asyncio.run(main())

gather 类似 Promise.all。它适合网络等待，不会自动加速 CPU 密集计算。

不要阻塞事件循环

async 函数里调用同步 SDK、time.sleep() 或很重的同步文件处理仍然会阻塞。网络 API 优先使用异步客户端；已有同步函数可以用 await asyncio.to_thread(fn, args) 转到线程执行。

捕获异常

JavaScript

try {
  JSON.parse('{bad json}')
} catch (error) {
  console.log(error.message)
} finally {
  console.log('结束')
}

Python

import json

try:
    json.loads("{bad json}")
except json.JSONDecodeError as error:
    print(error.msg)
finally:
    print("结束")

捕获尽量具体的异常：输入错误可以提示重试；权限错误不应该悄悄忽略；超时可以有限重试。不要用 except: pass 把所有失败藏起来。

读写文件：pathlib

import json
from pathlib import Path

path = Path("notes.json")
data = [{"title": "学习 Agent", "done": False}]
path.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8")
loaded = json.loads(path.read_text(encoding="utf-8"))
print(loaded[0]["title"])
print(path.resolve())

文件名相对当前工作目录解析，不一定是代码文件所在目录。想相对于代码位置：Path(__file__).resolve().parent。

with 是管理资源的语法，相当于「进入后使用，离开时自动清理」：

with open("notes.json", encoding="utf-8") as file:
    text = file.read()
# 离开 with 后文件自动关闭

调一个 HTTP 接口

uv add httpx

import asyncio
import httpx

async def main():
    async with httpx.AsyncClient(timeout=10) as client:
        response = await client.get("https://httpbin.org/get", params={"topic": "agent"})
        response.raise_for_status()  # 4xx / 5xx 转成异常
        print(response.json()["args"])

asyncio.run(main())

这是 fetch / axios 的位置。设置 timeout、检查 HTTP 状态、处理响应数据——和你做前端网络请求时的原则一致。示例需要网络，超时不等于 Python 配置失败。

动手练习

写一个 read_notes(path) 函数：文件不存在返回空列表；JSON 损坏时打印清晰提示并抛出异常，不要覆盖坏文件。

参考答案

import json
from pathlib import Path

def read_notes(path: Path) -> list:
    if not path.exists():
        return []
    try:
        return json.loads(path.read_text(encoding="utf-8"))
    except json.JSONDecodeError:
        print(f"{path} 不是有效 JSON，请先备份并修复")
        raise

下一课把这些知识串起来：命令行记账本。