如何检查文件对象是否支持随机访问?含DataLoader类应用场景
实现支持随机访问判断的DataLoader
下面是完善后的代码实现,我调整了随机访问的判断逻辑(遵循Python标准库的seekable()方法,而非自定义属性),并补全了延迟加载和全量加载的核心逻辑:
from collections import namedtuple # 定义Header类型,符合你提到的可哈希、带name/size/offset的要求 Header = namedtuple('Header', ['name', 'size', 'offset']) class DataLoader: def __init__(self, file): self.file = file self.headers = {} # 存储结构: {name: (header, data_or_None)} self._loaded_all = False # 标记是否已全量加载 def load_next_header(self): """从文件中读取下一个头部的实现示例,你可以根据实际格式修改""" # 假设头部是按固定格式读取:name长度(4字节) + size(4字节) + offset(8字节) try: name_len = int.from_bytes(self.file.read(4), byteorder='little') name = self.file.read(name_len).decode('utf-8') size = int.from_bytes(self.file.read(4), byteorder='little') offset = int.from_bytes(self.file.read(8), byteorder='little') return Header(name, size, offset) except ValueError: # 读取到不完整的头部,判定为文件结束 raise EOFError("End of file reached while reading header") def load_data(self): while True: try: header = self.load_next_header() except EOFError: # 读取到文件末尾时退出循环 break if self.file.seekable(): # 支持随机访问:只存头部,数据延迟加载 self.headers[header.name] = (header, None) else: # 不支持随机访问:立即加载对应数据 self.file.seek(header.offset) data = self.file.read(header.size) self.headers[header.name] = (header, data) if not self.file.seekable(): self._loaded_all = True def get_data(self, name): """根据名称获取数据,触发延迟加载(如果支持随机访问)""" if name not in self.headers: raise KeyError(f"Header '{name}' not found") header, data = self.headers[name] if data is None and self.file.seekable(): # 延迟加载数据 self.file.seek(header.offset) data = self.file.read(header.size) self.headers[name] = (header, data) return data
关键细节说明:
- 随机访问判断:用Python标准的
file.seekable()方法替代自定义random_access属性,这样能兼容所有标准可随机访问的文件对象(比如本地文件、io.BytesIO等)和不可随机访问的对象(比如管道、网络流)。 - 延迟加载逻辑:当文件支持随机访问时,
load_data()只解析并存储头部信息,数据会在第一次调用get_data()时才从指定偏移量读取,节省内存占用。 - 全量加载逻辑:如果文件不支持随机访问(比如流式输入),
load_data()会一次性把所有头部对应的数据都读取并存储,避免后续无法回溯读取。 - Header类型:用
namedtuple定义,满足可哈希、字段明确的要求,能直接作为字典的键值使用。
内容的提问来源于stack exchange,提问作者Mad Physicist
相关产品推荐
相关产品推荐

