FastAPI StreamingResponse返回Excel是否存在资源泄漏及如何优化

原实现问题说明 #

• 资源泄漏风险：原实现中额外创建的BytesIO属于内存对象，Python的垃圾回收机制会在对象引用清零后自动回收，不存在持久化资源泄漏问题，但确实存在两处可优化点：一是冗余的全量内存拷贝，二是不符合显式资源管理的最佳实践。
• 报错原因：你将StreamingResponse使用的字节流包裹在with块/contextlib.closing中时，with块会在路由函数return前执行退出逻辑关闭流，而FastAPI的StreamingResponse是在路由函数返回后才异步迭代读取流内容，此时操作已关闭的流就会触发ValueError: I/O operation on closed file.报错。

优化实现方案 #

方案1：极简无拷贝实现（适合绝大多数场景） #

直接复用写入用的buffer，无需额外创建新的字节流，也没有内存拷贝开销，内存对象由GC自动回收：

from fastapi.responses import StreamingResponse
...
buffer = BytesIO()
# pylint: disable=abstract-class-instantiated
with pandas.ExcelWriter(path=buffer, engine=XCL_ENGN) as writer:
    result1.to_excel(excel_writer=writer, sheet_name='Report 1', index=False)
    result2.to_excel(excel_writer=writer, sheet_name='Report 2', index=False)
    result3.to_excel(excel_writer=writer, sheet_name='Report 3', index=False)
# 将流指针移动到起始位置，直接传入StreamingResponse
buffer.seek(0)
return StreamingResponse(content=buffer, headers=headers)

方案2：显式资源管理实现（符合严格的上下文管理规范） #

如果需要显式控制资源释放，可以通过生成器包裹逻辑，确保流读取完成后自动关闭：

from fastapi.responses import StreamingResponse
from contextlib import contextmanager
...
@contextmanager
def managed_buffer():
    buffer = BytesIO()
    try:
        yield buffer
    finally:
        buffer.close()

def excel_stream():
    with managed_buffer() as buffer:
        with pandas.ExcelWriter(path=buffer, engine=XCL_ENGN) as writer:
            result1.to_excel(excel_writer=writer, sheet_name='Report 1', index=False)
            result2.to_excel(excel_writer=writer, sheet_name='Report 2', index=False)
            result3.to_excel(excel_writer=writer, sheet_name='Report 3', index=False)
        buffer.seek(0)
        # 迭代输出流内容
        yield from buffer

return StreamingResponse(content=excel_stream(), headers=headers)

该方案下，buffer会在StreamingResponse完全读取完流内容后，由上下文管理器的finally逻辑自动关闭，完全符合资源管理最佳实践，也没有冗余拷贝。

内容的提问来源于stack exchange，提问作者Della