Python 上下文管理器与文件操作：资源安全管理

本文适合谁：了解 Java try-with-resources 和 AutoCloseable 的工程师，想理解 Python with 语句和上下文管理器的工作原理。读完本篇，你能安全管理 AI 项目中的资源：HTTP 连接、临时文件、数据库连接。

文件句柄（打开文件后系统返回的操作凭证）、数据库连接、网络套接字（网络通信的端点）——这类资源有一个共同特点：用完必须释放。忘记关闭文件会导致文件描述符泄漏（系统能同时打开的文件数有上限）；忘记关闭数据库连接会耗尽连接池（预先建立的一批数据库连接，供多个请求复用）；在 LLM API 客户端的生命周期没有正确管理时，底层的 HTTP 连接池会持续占用。

Python 的上下文管理器（Context Manager）提供了一个统一的模式来解决这个问题：用 with 语句声明资源的生命周期，确保无论是否发生异常，退出时都会执行清理代码。

1.1 with 语句的底层原理

enter 到 exit 的完整生命周期，包含异常处理路径

with 语句背后是两个方法：__enter__ 和 __exit__。

python

# with 语句的语法糖展开：
with open("data.txt") as f:
    content = f.read()

# 等价于：
f = open("data.txt")
f.__enter__()          # 进入上下文，通常返回资源对象本身
try:
    content = f.read()
finally:
    f.__exit__(None, None, None)  # 无论是否异常，都会执行

__exit__ 的三个参数是异常信息：exc_type（异常类型）、exc_val（异常值）、exc_tb（调用栈）。没有异常时三个都是 None；返回 True 表示异常已处理，返回 False 或 None 表示让异常继续向上传播。

手动实现一个上下文管理器，直观理解执行顺序：

python

class Timer:
    """计时器上下文管理器：测量代码块耗时"""
    import time

    def __init__(self, label: str):
        self.label = label
        self.elapsed = 0.0

    def __enter__(self):
        self._start = self.time.monotonic()
        return self  # 返回 self，这样 'as timer' 拿到的是 Timer 实例

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.elapsed = self.time.monotonic() - self._start
        print(f"{self.label}: {self.elapsed:.3f}s")
        return False  # 不压制异常，异常会继续传播

with Timer("数据处理") as t:
    data = list(range(1_000_000))
    total = sum(data)

print(f"耗时 {t.elapsed:.3f}s，结果：{total}")
# 输出：数据处理: 0.045s
#       耗时 0.045s，结果：499999500000

1.2 contextlib.contextmanager：用生成器写上下文管理器

手动写 __enter__ / __exit__ 需要一个专门的类，代码量不少。contextlib.contextmanager 装饰器允许用一个生成器函数来定义上下文管理器：

python

from contextlib import contextmanager
import tempfile, shutil, os
from pathlib import Path

@contextmanager
def temp_directory(prefix: str = "ai_work_"):
    """
    创建临时工作目录，退出时自动清理
    yield 之前的代码 = __enter__
    yield 之后的代码 = __exit__
    """
    tmpdir = tempfile.mkdtemp(prefix=prefix)
    try:
        yield Path(tmpdir)  # 把临时目录路径交给 with 块使用
    finally:
        # 无论 with 块里发生什么，都会清理临时目录
        shutil.rmtree(tmpdir, ignore_errors=True)

# 使用示例：下载模型文件到临时目录，处理完自动删除
with temp_directory("model_download_") as work_dir:
    output_file = work_dir / "result.json"
    output_file.write_text('{"status": "done"}')
    print(f"临时目录：{work_dir}")
    # 做完工作后，with 块退出，临时目录被自动删除

# 此时 work_dir 已不存在
print(work_dir.exists())  # False

yield 前后的代码对应 __enter__ 和 __exit__，try/finally 保证清理代码必然执行。这个模式比写类简洁很多。

1.3 AI 开发中的典型场景

1.3.1 临时目录管理

AI 推理任务经常需要临时文件：缓存中间结果、保存临时图片、存放模型下载的权重文件。临时目录用完就删，是干净的工程实践：

python

from contextlib import contextmanager
import tempfile, shutil
from pathlib import Path

@contextmanager
def inference_workspace(model_name: str):
    """
    为单次推理任务创建独立的工作空间
    多个并发推理任务不会互相干扰
    """
    workspace = Path(tempfile.mkdtemp(prefix=f"{model_name}_"))
    (workspace / "inputs").mkdir()
    (workspace / "outputs").mkdir()
    try:
        yield workspace
    finally:
        shutil.rmtree(workspace, ignore_errors=True)

with inference_workspace("stable_diffusion") as ws:
    input_path  = ws / "inputs" / "prompt.txt"
    output_path = ws / "outputs" / "image.png"
    input_path.write_text("a cat on the moon")
    # run_inference(input_path, output_path)
    print(f"输出：{output_path}")

1.3.2 数据库连接池

python

from contextlib import contextmanager
from typing import Generator
import sqlite3

class DatabasePool:
    """简化的连接池，演示上下文管理器在数据库场景的应用"""

    def __init__(self, db_path: str):
        self.db_path = db_path

    @contextmanager
    def get_connection(self) -> Generator[sqlite3.Connection, None, None]:
        """
        获取连接 -> 使用 -> 自动提交/回滚 -> 归还连接
        事务管理和资源归还都在这里处理，调用方无需关心
        """
        conn = sqlite3.connect(self.db_path)
        conn.row_factory = sqlite3.Row  # 让查询结果可以用列名访问
        try:
            yield conn
            conn.commit()    # with 块正常结束则提交
        except Exception:
            conn.rollback()  # with 块抛出异常则回滚
            raise            # 让异常继续传播，不在这里静默吞掉
        finally:
            conn.close()     # 无论如何都关闭连接

pool = DatabasePool("vectors.db")

with pool.get_connection() as conn:
    conn.execute("""
        CREATE TABLE IF NOT EXISTS embeddings
        (id TEXT PRIMARY KEY, vector BLOB, metadata TEXT)
    """)
    # 如果这里抛出异常，事务自动回滚，连接自动关闭

1.3.3 API 客户端生命周期

python

from contextlib import contextmanager
from openai import OpenAI
import httpx

@contextmanager
def managed_openai_client(timeout: float = 30.0):
    """
    管理 OpenAI 客户端的生命周期
    确保底层 HTTP 连接池被正确关闭，避免资源泄漏
    """
    # 自定义 HTTP 客户端，精确控制超时和连接池大小
    http_client = httpx.Client(
        timeout=httpx.Timeout(timeout),
        limits=httpx.Limits(max_connections=10)
    )
    client = OpenAI(http_client=http_client)
    try:
        yield client
    finally:
        http_client.close()  # 关闭连接池，释放所有底层 TCP 连接

with managed_openai_client(timeout=60.0) as client:
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": "Hello"}]
    )
    print(response.choices[0].message.content)

1.4 pathlib：现代文件路径操作

Python 3.4+ 的 pathlib 用面向对象的方式操作文件路径，替代了旧的字符串拼接和 os.path 函数。

python

from pathlib import Path

# 路径拼接：用 / 运算符，比 os.path.join 更直观
project_root = Path("/Users/alice/projects/ai_app")
data_dir     = project_root / "data"
config_file  = project_root / "config" / "settings.yaml"

# 常用属性
print(config_file.name)      # "settings.yaml"
print(config_file.stem)      # "settings"（不含扩展名）
print(config_file.suffix)    # ".yaml"
print(config_file.parent)    # /Users/alice/projects/ai_app/config

# 创建目录（包括中间目录）
(project_root / "logs" / "2024").mkdir(parents=True, exist_ok=True)

# 文件读写：比 open() 更简洁
config_file.parent.mkdir(parents=True, exist_ok=True)
config_file.write_text("model: gpt-4o\ntemperature: 0.7", encoding="utf-8")
content = config_file.read_text(encoding="utf-8")

# 遍历目录
for py_file in project_root.rglob("*.py"):  # rglob 是递归 glob
    print(py_file.relative_to(project_root))

# 检查文件状态
if config_file.exists() and config_file.is_file():
    size_kb = config_file.stat().st_size / 1024
    print(f"配置文件大小：{size_kb:.1f} KB")

与 Java Path 的对比：

操作	Python pathlib	Java NIO Path
路径拼接	`Path("a") / "b" / "c"`	`Paths.get("a").resolve("b").resolve("c")`
读取文本	`path.read_text()`	`Files.readString(path)`
写入文本	`path.write_text(s)`	`Files.writeString(path, s)`
创建目录	`path.mkdir(parents=True)`	`Files.createDirectories(path)`
递归遍历	`path.rglob("*.py")`	`Files.walk(path)`
文件大小	`path.stat().st_size`	`Files.size(path)`

1.5 JSON/YAML 配置文件读写

AI 项目的配置通常存在 JSON 或 YAML 文件里，包括模型参数、提示词模板、服务地址等。

python

import json
from pathlib import Path

def load_config(config_path: str | Path) -> dict:
    """加载 JSON 配置，文件不存在时返回默认值"""
    path = Path(config_path)
    if not path.exists():
        return {}
    with open(path, encoding="utf-8") as f:
        return json.load(f)

def save_config(config: dict, config_path: str | Path) -> None:
    """保存配置到 JSON 文件，自动创建父目录"""
    path = Path(config_path)
    path.parent.mkdir(parents=True, exist_ok=True)
    with open(path, "w", encoding="utf-8") as f:
        # ensure_ascii=False：中文直接写入，不转义为 \uXXXX
        # indent=2：格式化输出，方便人工查看
        json.dump(config, f, ensure_ascii=False, indent=2)

# YAML 配置（需要 pip install pyyaml）
import yaml

def load_yaml_config(path: str | Path) -> dict:
    """加载 YAML 配置文件"""
    with open(path, encoding="utf-8") as f:
        return yaml.safe_load(f)  # safe_load 不执行任意 Python 代码，安全

# 典型 AI 项目配置文件结构
config = {
    "model": {
        "name": "gpt-4o",
        "temperature": 0.7,
        "max_tokens": 2048,
    },
    "prompts": {
        "system": "你是一个专业的代码助手",
        "user_template": "请分析以下代码：{code}",
    },
    "rate_limits": {
        "requests_per_minute": 60,
        "tokens_per_minute": 100000,
    }
}

save_config(config, "config/app_config.json")
loaded = load_config("config/app_config.json")
print(loaded["model"]["name"])  # "gpt-4o"

1.6 环境变量管理

API Key、数据库密码这类敏感信息绝不能写进代码或配置文件。标准做法是用环境变量（操作系统级别的键值对配置，程序启动时读取，不存放在代码文件里），python-dotenv 库让本地开发时的环境变量管理更方便：

python

# .env 文件（加入 .gitignore，绝不提交到版本库）
# OPENAI_API_KEY=sk-proj-...
# DEEPSEEK_API_KEY=your_deepseek_api_key_here
# DATABASE_URL=postgresql://user:pass@localhost/mydb
# DEBUG=true

from dotenv import load_dotenv
import os

# 加载 .env 文件到环境变量（生产环境不需要，环境变量已由 k8s/docker 注入）
# k8s（Kubernetes）和 docker 是常用的容器化部署工具，它们可以在启动应用时自动注入环境变量
load_dotenv()

# 读取环境变量，提供类型转换和默认值
OPENAI_API_KEY = os.environ["OPENAI_API_KEY"]      # 不存在则抛 KeyError
DEBUG          = os.getenv("DEBUG", "false") == "true"  # 有默认值
MAX_RETRIES    = int(os.getenv("MAX_RETRIES", "3"))     # 类型转换

# 推荐做法：集中管理配置，避免散落在各处的 os.getenv()
from dataclasses import dataclass

@dataclass
class AppConfig:
    """从环境变量加载应用配置，在启动时集中校验"""
    openai_api_key: str
    debug: bool
    max_retries: int
    log_level: str

    @classmethod
    def from_env(cls) -> "AppConfig":
        """工厂方法：从环境变量创建配置对象，缺少必须项则早失败"""
        return cls(
            openai_api_key = os.environ["OPENAI_API_KEY"],
            debug          = os.getenv("DEBUG", "false").lower() == "true",
            max_retries    = int(os.getenv("MAX_RETRIES", "3")),
            log_level      = os.getenv("LOG_LEVEL", "INFO").upper(),
        )

config = AppConfig.from_env()

完整的上下文管理器工作流：

1.7 AI 项目资源管理的典型架构

在一个完整的 AI 应用中，资源管理涉及多个层次：

python

import asyncio
from contextlib import asynccontextmanager
from typing import AsyncIterator
import httpx
from openai import AsyncOpenAI

# AI 项目的完整资源管理示例
# 展示从应用启动到关闭，所有资源的生命周期管理

class AIAppResources:
    """集中管理 AI 应用所有资源的生命周期"""

    def __init__(self):
        self._http_client: httpx.AsyncClient | None = None
        self._llm_client: AsyncOpenAI | None = None
        self._db_pool = None  # 数据库连接池

    @asynccontextmanager
    async def lifespan(self) -> AsyncIterator["AIAppResources"]:
        """
        整个应用的生命周期管理。
        FastAPI 的 lifespan 钩子就是这个模式。
        """
        # === 启动阶段：初始化所有资源 ===
        print("初始化 HTTP 客户端...")
        self._http_client = httpx.AsyncClient(
            timeout=httpx.Timeout(connect=5.0, read=120.0),
            limits=httpx.Limits(max_connections=50)
        )

        print("初始化 LLM 客户端...")
        self._llm_client = AsyncOpenAI(http_client=self._http_client)

        print("连接数据库...")
        # self._db_pool = await create_pool(DATABASE_URL)

        print("所有资源初始化完成，应用就绪。")

        try:
            yield self  # 把资源对象提供给应用使用
        finally:
            # === 关闭阶段：按顺序释放资源 ===
            print("关闭数据库连接池...")
            # if self._db_pool:
            #     await self._db_pool.close()

            print("关闭 HTTP 连接池...")
            if self._http_client:
                await self._http_client.aclose()

            print("所有资源已释放，应用安全退出。")

    @property
    def llm(self) -> AsyncOpenAI:
        if self._llm_client is None:
            raise RuntimeError("资源未初始化，请先进入 lifespan 上下文")
        return self._llm_client


# FastAPI 中的实际用法
from fastapi import FastAPI

app_resources = AIAppResources()

# FastAPI 的 lifespan 参数管理启动/关闭
@asynccontextmanager
async def lifespan(app: FastAPI):
    async with app_resources.lifespan():
        yield

app = FastAPI(lifespan=lifespan)

@app.post("/chat")
async def chat(message: str) -> dict:
    response = await app_resources.llm.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": message}]
    )
    return {"reply": response.choices[0].message.content}

与 Java Spring 的生命周期对比：

Java Spring	Python FastAPI + 上下文管理器
`@PostConstruct`	`lifespan` 进入时的初始化代码
`@PreDestroy`	`lifespan` 退出时的清理代码（`finally` 块）
`@Bean` 单例管理	模块级单例（`app_resources = AIAppResources()`）
`@Scope("prototype")`	每次请求新建（在请求处理函数内 `with ...`）

1.8 小结

上下文管理器是 Python 资源管理的核心模式。with 语句的本质是一对生命周期钩子：进入时获取资源，退出时释放资源，finally 语义保证清理代码必然执行。

contextlib.contextmanager 装饰器让定义上下文管理器变得和写普通函数一样简单，yield 之前是初始化，yield 值是给 as 子句使用的对象，yield 之后（在 finally 块里）是清理逻辑。

pathlib 是文件操作的现代写法，用 / 拼接路径、直接调用 read_text()/write_text() 比 os.path 和裸 open() 简洁很多。环境变量管理要遵循"敏感信息不进代码库"的原则，python-dotenv 是本地开发的标配工具。

下一篇将介绍日志与调试，如何在 AI 应用中配置结构化日志、记录 LLM 调用的详细信息，以及如何调试复杂的异步代码。