Python获取内存中数据行数及实现支持索引与len的自定义类

内存中行数统计的注意事项 #

你现在的实现是一次性将所有文件内容加载到内存中处理，需要注意以下和内存行数相关的核心问题：

• 全量加载的内存成本：如果处理GB级别的大文件，readlines会一次性把所有行读入内存，直接导致内存占用过高甚至程序崩溃。此时内存中存储的行数就是self.__data列表的长度，如果存在空行、格式错误的行，解析时如果做了跳过处理，内存行数会和原文件实际行数不一致，需要提前明确len()的统计规则：是统计原文件所有行，还是统计内存中有效解析后的行。
• 惰性加载的一致性问题：如果后续优化为懒加载模式（即读一行处理一行，不用一次性加载全量数据），不能直接用内存中已加载的行数作为len()的返回值，否则会出现用户前几次调用len()返回已加载行数、全量加载完成后才返回总行数的前后不一致问题，需要在初始化时先遍历一遍文件统计原文件总行数单独存储。
• 数据修改的同步规则：如果后续支持对已加载的行做增删改操作，必须保证__len__返回值和内存中实际存储的行数、记录的原文件行数同步，避免出现计数偏差。

现有实现优化 #

你当前的代码已经基本满足「可索引、len()返回行数」的需求，仅存在少量逻辑缺陷可以优化：

1. 索引判断存在bug：Python索引从0开始，当前if index > len(self.__data)的判断会漏掉index == len(self.__data)的越界场景，同时未支持切片索引（比如data[1:3]的用法）。
2. 无解析容错逻辑：如果某行JSON格式错误，json.loads会直接抛出异常导致类初始化失败，可增加异常捕获逻辑兼容无效行。

优化后的代码如下：

import json
import re

class DataLoader(object):
    
    def __init__(self, file_name, skip_invalid_rows: bool = True):
        self.skip_invalid = skip_invalid_rows
        # 存储原文件总行数，适配需要返回原文件行数的需求
        self.raw_file_line_count = 0
        self.__data = []
        with open(file_name, 'r', encoding='utf-8') as file:
            for row in file:
                self.raw_file_line_count += 1
                processed_row = self.__replace(row.strip())
                if not processed_row:
                    continue
                try:
                    parsed_row = json.loads(processed_row)
                    self.__data.append(parsed_row)
                except json.JSONDecodeError as e:
                    if not self.skip_invalid:
                        raise ValueError(f"第{self.raw_file_line_count}行解析失败：{str(e)}") from e
    
    def __getitem__(self, index):
        # 同时支持整数索引和切片索引
        if isinstance(index, (int, slice)):
            return self.__data[index]
        raise TypeError("Indices must be integers or slices")
                
    def __len__(self):
        # 需要返回原文件总行数则修改为返回 self.raw_file_line_count
        # 需要返回内存中有效解析行数则保持返回 len(self.__data)
        return len(self.__data)
  
    def __replace(self,row):
        # 辅助函数，将'_i'替换为'index'
        return re.sub(r'_i','index',row)

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