python-resource-management

来源声明：本技能内容源自 wshobson/agents （MIT License, Copyright (c) 2024 Seth Hobson；完整许可文本见同目录 LICENSE.txt）。 使用、修改与再分发须遵守该 MIT 许可条款并保留版权声明。

中文描述

本技能提供全栈开发相关的最佳实践和模式指导。

---

name: python-resource-management description: 使用上下文管理器、清理模式和流式处理来管理 Python 资源。在管理连接、文件句柄、实现清理逻辑或构建带累积状态的流式响应时使用。

Python 资源管理

使用上下文管理器确定性地管理资源。数据库连接、文件句柄和网络套接字等资源应可靠释放，即使发生异常也不例外。

何时使用此技能

• 管理数据库连接和连接池
• 处理文件句柄和 I/O
• 实现自定义上下文管理器
• 构建带状态的流式响应
• 处理嵌套资源清理
• 创建异步上下文管理器

核心概念

1. 上下文管理器

with 语句可确保资源自动释放，即使发生异常也不例外。

2. 协议方法

同步资源管理使用 __enter__/__exit__，异步资源管理使用 __aenter__/__aexit__。

3. 无条件清理

无论是否发生异常，__exit__ 都会运行。

4. 异常处理

从 __exit__ 返回 True 可抑制异常，返回 False 则会传播异常。

快速入门

from contextlib import contextmanager

@contextmanager
def managed_resource():
    resource = acquire_resource()
    try:
        yield resource
    finally:
        resource.cleanup()

with managed_resource() as r:
    r.do_work()

基本模式

模式 1：基于类的上下文管理器

为复杂资源实现上下文管理器协议。

class DatabaseConnection:
    """Database connection with automatic cleanup."""

    def __init__(self, dsn: str) -> None:
        self._dsn = dsn
        self._conn: Connection | None = None

    def connect(self) -> None:
        """Establish database connection."""
        self._conn = psycopg.connect(self._dsn)

    def close(self) -> None:
        """Close connection if open."""
        if self._conn is not None:
            self._conn.close()
            self._conn = None

    def __enter__(self) -> "DatabaseConnection":
        """Enter context: connect and return self."""
        self.connect()
        return self

    def __exit__(
        self,
        exc_type: type[BaseException] | None,
        exc_val: BaseException | None,
        exc_tb: TracebackType | None,
    ) -> None:
        """Exit context: always close connection."""
        self.close()

# Usage with context manager (preferred)
with DatabaseConnection(dsn) as db:
    result = db.execute(query)

# Manual management when needed
db = DatabaseConnection(dsn)
db.connect()
try:
    result = db.execute(query)
finally:
    db.close()

模式 2：异步上下文管理器

对于异步资源，实现异步协议。

class AsyncDatabasePool:
    """Async database connection pool."""

    def __init__(self, dsn: str, min_size: int = 1, max_size: int = 10) -> None:
        self._dsn = dsn
        self._min_size = min_size
        self._max_size = max_size
        self._pool: asyncpg.Pool | None = None

    async def __aenter__(self) -> "AsyncDatabasePool":
        """Create connection pool."""
        self._pool = await asyncpg.create_pool(
            self._dsn,
            min_size=self._min_size,
            max_size=self._max_size,
        )
        return self

    async def __aexit__(
        self,
        exc_type: type[BaseException] | None,
        exc_val: BaseException | None,
        exc_tb: TracebackType | None,
    ) -> None:
        """Close all connections in pool."""
        if self._pool is not None:
            await self._pool.close()

    async def execute(self, query: str, *args) -> list[dict]:
        """Execute query using pooled connection."""
        async with self._pool.acquire() as conn:
            return await conn.fetch(query, *args)

# Usage
async with AsyncDatabasePool(dsn) as pool:
    users = await pool.execute("SELECT * FROM users WHERE active = $1", True)

模式 3：使用 @contextmanager 装饰器

对于简单场景，使用装饰器简化上下文管理器。

from contextlib import contextmanager, asynccontextmanager
import time
import structlog

logger = structlog.get_logger()

@contextmanager
def timed_block(name: str):
    """Time a block of code."""
    start = time.perf_counter()
    try:
        yield
    finally:
        elapsed = time.perf_counter() - start
        logger.info(f"{name} completed", duration_seconds=round(elapsed, 3))

# Usage
with timed_block("data_processing"):
    process_large_dataset()

@asynccontextmanager
async def database_transaction(conn: AsyncConnection):
    """Manage database transaction."""
    await conn.execute("BEGIN")
    try:
        yield conn
        await conn.execute("COMMIT")
    except Exception:
        await conn.execute("ROLLBACK")
        raise

# Usage
async with database_transaction(conn) as tx:
    await tx.execute("INSERT INTO users ...")
    await tx.execute("INSERT INTO audit_log ...")

模式 4：无条件释放资源

无论是否发生异常，都要在 __exit__ 中清理资源。

class FileProcessor:
    """Process file with guaranteed cleanup."""

    def __init__(self, path: str) -> None:
        self._path = path
        self._file: IO | None = None
        self._temp_files: list[Path] = []

    def __enter__(self) -> "FileProcessor":
        self._file = open(self._path, "r")
        return self

    def __exit__(
        self,
        exc_type: type[BaseException] | None,
        exc_val: BaseException | None,
        exc_tb: TracebackType | None,
    ) -> None:
        """Clean up all resources unconditionally."""
        # Close main file
        if self._file is not None:
            self._file.close()

        # Clean up any temporary files
        for temp_file in self._temp_files:
            try:
                temp_file.unlink()
            except OSError:
                pass  # Best effort cleanup

        # Return None/False to propagate any exception

详细示例和模式

详细章节（从 ## Advanced Patterns 开始）位于 references/details.md。如果上面的导航摘要不足以满足需要，请阅读该文件。

最佳实践总结

1. 始终使用上下文管理器 - 对任何需要清理的资源都应如此
2. 无条件清理 - 即使发生异常也会运行 __exit__
3. 不要意外抑制异常 - 除非有意抑制，否则返回 False
4. 使用 @contextmanager - 适用于简单的资源管理模式
5. 实现两种协议 - 支持 with 和手动管理
6. 使用 ExitStack - 用于管理数量动态变化的资源
7. 高效累积内容 - 使用列表加 join，而不是字符串拼接
8. 跟踪指标 - 对流式处理而言，首字节时间很重要
9. 记录行为 - 尤其要说明异常抑制行为
10. 测试清理路径 - 验证发生错误时资源是否得到释放

能力边界

✅ 擅长处理

• 全栈开发相关的最佳实践
• 代码架构设计指导
• 性能优化建议

⚠️ 需要素材

• 具体的项目上下文
• 技术栈信息

❌ 超出范围

• 非技术领域的业务决策
• 具体的代码实现（仅提供指导）

工作流程

第 1 步：需求分析

理解项目需求和技术约束

第 2 步：架构设计

设计系统架构和组件划分

第 3 步：实现指导

提供具体的实现建议和代码模式

第 4 步：代码审查

审查代码质量和最佳实践遵循情况

第 5 步：优化建议

提供性能优化和改进建议

注意事项

1. 过度设计

避免在简单场景中使用复杂的设计模式

2. 忽视性能

不要忽视性能优化，特别是在生产环境中

3. 安全考量

始终考虑安全性和数据保护

4. 测试覆盖率

确保有足够的测试覆盖，特别是边界情况

5. 文档维护

保持文档与代码同步更新