工厂模式下文件对象资源管理：如何将文件所有权从工厂转移至解析器

解决方案 #

核心思路是取消工厂层对文件句柄的自动关闭逻辑，将检测格式时打开的文件句柄直接移交到匹配的解析器实例，由解析器根据使用场景管理生命周期，即可避免重复打开文件。

具体修改点 #

1. 工厂层不再用with自动管理文件句柄，改为手动打开，匹配到解析器后直接移交所有权，匹配失败则手动关闭句柄避免资源泄漏
2. 解析器的上下文管理器优先使用已持有的打开句柄，不存在时再重新打开
3. 补充非上下文使用场景下的句柄释放逻辑，避免资源泄漏

修改后完整代码 #

import io

class OpenWrapper:
    # 用于演示文件打开关闭状态的包装类
    def __init__(self, filename:io.BytesIO, mode) -> None:
        self.file = filename
        self.mode = mode
        self.instance = None

    def __enter__(self):
        print("Opening File")
        return self

    def __exit__(self, type, value, traceback):
        print("Closing File")

class FormatA:
    def __init__(self, filename, file_handle:OpenWrapper):
        # 检测文件格式是否匹配
        file_handle.file.seek(0)
        expected_header = "FormatA".encode()
        header = file_handle.file.read(len(expected_header))
        if header != expected_header:
            raise RuntimeError("Can not parse this file.")

        self.filename = filename
        # 接收工厂移交的打开状态的句柄
        self.file_handler:OpenWrapper = file_handle
        # 标记句柄是否被上下文管理器接管
        self._managed_by_context = False

    def read(self):
        if self.file_handler is not None:
            # 跳过文件头
            self.file_handler.file.seek(8)
            data = self.file_handler.file.read().decode("utf-8")
            # 如果不是上下文管理器接管的场景，读完直接关闭句柄
            if not self._managed_by_context:
                self.file_handler.__exit__(None, None, None)
                self.file_handler = None
        else:
            # 没有可用句柄时重新打开
            with OpenWrapper(self.filename, "rb") as file_handle:
                file_handle.file.seek(8)
                data = file_handle.file.read().decode("utf-8")

        return data.split("_")

    def __enter__(self):
        self._managed_by_context = True
        # 已有打开的句柄直接使用，不需要重复打开
        if self.file_handler is None:
            self.file_handler = OpenWrapper(self.filename, "rb").__enter__()
        return self

    def __exit__(self, type, value, traceback):
        if self.file_handler is not None:
            self.file_handler.__exit__(type, value, traceback)
            self.file_handler = None

class ParserFactory:
    def __init__(self):
        self.known_plugins = [FormatA]

    def __call__(self, filename):
        mode = "rb" 
        # 手动打开句柄，不使用with自动关闭
        file_handle = OpenWrapper(filename, mode).__enter__()
        try:
            for plugin_class in self.known_plugins:
                try:
                    plugin = plugin_class(filename, file_handle)
                    # 匹配到解析器，直接移交句柄所有权，工厂不再管理
                    return plugin
                except RuntimeError:
                    # 格式不匹配，重置文件指针试下一个插件
                    file_handle.file.seek(0)
                    continue
            # 所有插件都不匹配，手动关闭句柄
            file_handle.__exit__(None, None, None)
            raise RuntimeError("Could not find parser.")
        except Exception as e:
            # 异常场景下关闭句柄避免泄漏
            if file_handle is not None:
                file_handle.__exit__(None, None, None)
            raise e

fancy_loader = ParserFactory()

file1 = io.BytesIO("FormatA\nDATA_DATA_DATA_DATA".encode())
file2 = io.BytesIO("UNIQUE_FORMAT_B_HEADERDATA_DATA_DATA".encode())

# 场景1：使用上下文管理器
print("Scenario 1")
print("==========")
with fancy_loader(file1) as data_file:
    print(data_file.read())

# 场景2：直接调用read
print("\n")
print("Scenario 2")
print("==========")
# 重置文件指针，避免前一次读取影响格式检测
file1.seek(0)
data_file = fancy_loader(file1)
print(data_file.read())

运行输出 #

Scenario 1
==========
Opening File
['DATA', 'DATA', 'DATA', 'DATA']
Closing File


Scenario 2
==========
Opening File
Closing File
['DATA', 'DATA', 'DATA', 'DATA']

完全符合你期望的运行流程。

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