Python 文件操作进阶:pathlib、大文件流式读取与函数设计

Alice
Alice
发布时间:2026-10-13 | 3人浏览过
原创

本文深入探讨 Python 文件处理的三种进阶技巧。首先推荐使用 pathlib 模块简化路径操作并兼容旧接口;其次演示如何通过 iter 函数结合生成器实现大文件的高效分块读取,显著降低内存占用;最后讲解如何通过依赖“类文件对象”提升函数的可测试性与通用性,适用于 StringIO 和 subprocess 等多种场景。

Python 文件操作概述

在 Python 开发中,文件操作是日常任务的核心部分。无论是生成复杂的报表,还是快速解析整理数以万计的数据文件,掌握高效的文件处理技巧都能显著提升工作效率。

编写文件相关代码时,开发者通常关注执行速度与任务完成的效率。本文将分享三个实用的编程建议:介绍一个常被忽视的标准库模块,演示大文件读取的最佳实践,并探讨函数设计的优化思路。

接下来,让我们开始第一个模块推荐。

注意:由于不同操作系统的文件系统存在差异,本文主要基于 Mac OS/Linux 环境编写,部分代码在 Windows 系统上可能不适用。

使用 pathlib 模块简化路径处理

建议一:使用 pathlib 模块

在 Python 中进行文件处理时,标准库中的 os 和 os.path 模块几乎是必经之路。它们提供了大量处理文件路径、读写文件及查看文件状态的工具函数。

举例来说,假设有一个目录包含大量数据文件,后缀名不统一,既有 .txt 也有 .csv。我们需要将所有以 .txt 结尾的文件重命名为 .csv 后缀。

我们可以编写如下函数来实现这一需求:


  import os

  import os.path



  def unify_ext_with_os_path(path):

    """统一目录下的 .txt 文件名后缀为 .csv

    """

    for filename in os.listdir(path):

      basename, ext = os.path.splitext(filename)

      if ext == '.txt':

        abs_filepath = os.path.join(path, filename)

        os.rename(abs_filepath, os.path.join(path, f'{basename}.csv'))

上述代码涉及了多个文件处理函数,具体包括:

  • os.listdir(path):列出指定路径下的所有文件和子目录
  • os.path.splitext(filename):分离文件名的基本名称和后缀
  • os.path.join(path,filename):拼接路径与文件名生成绝对路径
  • os.rename(...):执行文件重命名操作

尽管上述函数能完成任务,但即便对于经验丰富的 Python 开发者而言,这些函数不仅难以记忆,生成的代码可读性也较差。

使用 pathlib 模块改写代码

为了简化文件处理,Python 3.4 引入了 pathlib 标准库模块。该模块基于面向对象设计,封装了丰富的文件操作功能。使用它重写上述代码,效果将截然不同。

使用 pathlib 模块后的代码如下:


  from pathlib import Path


  def unify_ext_with_pathlib(path):

    for fpath in Path(path).glob('*.txt'):

      fpath.rename(fpath.with_suffix('.csv'))

与旧代码相比,新代码仅需两行即可完成工作,主要步骤如下:

  1. 首先使用 Path(path) 将字符串路径转换为 Path 对象
  2. 调用 .glob('*.txt') 对路径内容进行模式匹配,以生成器方式返回 Path 对象,便于后续操作
  3. 使用 .with_suffix('.csv') 直接获取带有新后缀的文件全路径
  4. 调用 .rename(target) 完成重命名操作

相比 os 和 os.path,pathlib 模块使代码更加精简且风格统一,实现了文件操作的一站式处理。

其他用法

此外,pathlib 还提供了许多便捷用法,例如使用 / 运算符组合文件路径:


  # 旧朋友:使用 os.path 模块

  >>> import os.path

  >>> os.path.join('/tmp', 'foo.txt')

  '/tmp/foo.txt'


  # ✨ 新潮流:使用 / 运算符

  >>> from pathlib import Path

  >>> Path('/tmp') / 'foo.txt'

  PosixPath('/tmp/foo.txt')

或者使用 .read_text() 快速读取文件内容:


  # 标准做法,使用 with open(...) 打开文件

  >>> with open('foo.txt') as file:

  ...   print(file.read())

  ...

  foo


  # 使用 pathlib 可以让这件事情变得更简单

  >>> from pathlib import Path

  >>> print(Path('foo.txt').read_text())

  foo

除了文中介绍的功能,pathlib 模块还提供了大量实用方法,建议查阅官方文档深入了解。

如果上述功能仍不足以打动你,还有一个使用 pathlib 的理由:PEP-519 定义了专门用于“文件路径”的新对象协议。这意味着从 Python 3.6 版本起,pathlib 的 Path 对象可与绝大多数仅接受字符串路径的标准库函数兼容使用:


  >>> p = Path('/tmp')

  # 可以直接对 Path 类型对象 p 进行 join

  >>> os.path.join(p, 'foo.txt')

  '/tmp/foo.txt'

因此,无需犹豫,立即开始使用 pathlib 模块吧。

提示:如果你使用的是更早的 Python 版本,可以尝试安装 pathlib2 模块。

大文件流式读取与内存优化

建议二:掌握如何流式读取大文件

在 Python 中读取文件的“标准做法”通常是:使用 with open(file_name) 上下文管理器获取文件对象,然后通过 for 循环迭代文件,逐行获取内容。

以下是一个使用这种“标准做法”的简单示例函数:


  def count_nine(fname):

    """计算文件里包含多少个数字 '9'

    """

    count = 0

    with open(fname) as file:

      for line in file:

        count += line.count('9')

    return count

假设我们有一个文件 small_file.txt,使用该函数可以轻松统计其中数字 9 的数量。


  # small_file.txt

  feiowe9322nasd9233rl

  aoeijfiowejf8322kaf9a


  # OUTPUT: 3

  print(count_nine('small_file.txt'))

这种读取方式成为标准的原因在于其两大优势:

  • with 上下文管理器会自动关闭打开的文件描述符
  • 迭代文件对象时,内容逐行返回,不会占用过多内存

标准做法的缺点

然而,这套标准做法并非没有缺陷。如果被读取的文件中没有任何换行符,上述第二个优势将不再成立。当代码执行到 for line in file 时,line 将变成一个巨大的字符串对象,消耗大量内存。

让我们做一个实验:假设有一个 5GB 大的文件 big_file.txt,其中包含与 small_file.txt 相同的随机字符串,但所有文本都存储在同一行中:


# FILE: big_file.txt
df2if283rkwefh... <剩余 5GB 大小> ...

如果我们继续使用前面的 count_nine 函数统计这个大文件中 9 的数量,在我的笔记本上,该过程耗时 65 秒,并消耗了 2GB 内存 [注1]。

使用 read 方法分块读取

为解决此问题,我们需要暂时搁置“标准做法”,改用更底层的 file.read() 方法。与直接迭代文件对象不同,每次调用 file.read(chunk_size) 会直接返回从当前位置往后读取 chunk_size 大小的文件内容,无需等待换行符出现。

因此,如果使用 file.read() 方法,我们的函数可以改写为:


  def count_nine_v2(fname):

    """计算文件里包含多少个数字 '9',每次读取 8kb

    """

    count = 0

    block_size = 1024 * 8

    with open(fname) as fp:

      while True:

        chunk = fp.read(block_size)

        # 当文件没有更多内容时,read 调用将会返回空字符串 ''

        if not chunk:

          break

        count += chunk.count('9')

    return count

在新函数中,我们使用 while 循环读取文件内容,每次最多读取 8kb 大小,从而避免拼接巨大字符串的过程,大幅降低内存占用。

利用生成器解耦代码

如果讨论的不是 Python,上述代码或许已足够优秀。但仔细分析 count_nine_v2 函数,你会发现循环体内存在两个独立逻辑:数据生成(read 调用与 chunk 判断)与数据消费。这两个逻辑被耦合在一起。

正如我在《编写地道循环》中提到的,为提升复用能力,我们可以定义一个新的 chunked_file_reader 生成器函数,负责所有“数据生成”相关的逻辑。这样 count_nine_v3 中的主循环只需负责计数即可。


  def chunked_file_reader(fp, block_size=1024 * 8):

    """生成器函数:分块读取文件内容

    """

    while True:

      chunk = fp.read(block_size)

      # 当文件没有更多内容时,read 调用将会返回空字符串 ''

      if not chunk:

        break

      yield chunk



  def count_nine_v3(fname):

    count = 0

    with open(fname) as fp:

      for chunk in chunked_file_reader(fp):

        count += chunk.count('9')

    return count

至此,代码似乎已无优化空间,但并非如此。iter(iterable) 是构造迭代器的内建函数,但它还有一个鲜为人知的用法。当我们使用 iter(callable,sentinel) 方式调用它时,会返回一个特殊对象,迭代该对象将不断产生可调用对象 callable 的调用结果,直到结果为 sentinel 时迭代终止。


  def chunked_file_reader(file, block_size=1024 * 8):

    """生成器函数:分块读取文件内容,使用 iter 函数

    """

    # 首先使用 partial(fp.read, block_size) 构造一个新的无需参数的函数

    # 循环将不断返回 fp.read(block_size) 调用结果,直到其为 '' 时终止

    for chunk in iter(partial(file.read, block_size), ''):

      yield chunk

最终,仅需两行代码,我们就完成了一个可复用的分块文件读取函数。那么,该函数在性能方面的表现如何呢?

与最初的 2GB 内存/耗时 65 秒相比,使用生成器的版本仅需 7MB 内存 / 12 秒即可完成计算。效率提升了近 4 倍,内存占用更是不到原来的 1%。

设计接受文件对象的函数

建议三:设计接受文件对象的函数

统计完文件中的“9”之后,让我们换个需求。现在,我想统计每个文件中出现了多少个英文元音字母(aeiou)。只需对之前的代码稍作调整,即可很快写出新函数 count_vowels。


  def count_vowels(filename):

    """统计某个文件中,包含元音字母(aeiou)的数量

    """

    VOWELS_LETTERS = {'a', 'e', 'i', 'o', 'u'}

    count = 0

    with open(filename, 'r') as fp:

      for line in fp:

        for char in line:

          if char.lower() in VOWELS_LETTERS:

            count += 1

    return count



  # OUTPUT: 16

  print(count_vowels('small_file.txt'))

与之前“统计 9”的函数相比,新函数变得稍复杂。为保证程序正确性,我需要为其编写单元测试,但发现这非常麻烦,主要问题点如下:

  1. 函数接收文件路径作为参数,因此需要传递一个实际存在的文件
  2. 为准备测试用例,我必须提供几个样板文件或生成临时文件
  3. 文件是否能被正常打开、读取,也成了需要测试的边界情况

如果发现你的函数难以编写单元测试,通常意味着应该改进其设计。上述函数应如何改进?答案是:让函数依赖“文件对象”而非文件路径。

修改后的函数代码如下:


  def count_vowels_v2(fp):

    """统计某个文件中,包含元音字母(aeiou)的数量

    """

    VOWELS_LETTERS = {'a', 'e', 'i', 'o', 'u'}

    count = 0

    for line in fp:

      for char in line:

        if char.lower() in VOWELS_LETTERS:

          count += 1

    return count



  # 修改函数后,打开文件的职责被移交给了上层函数调用者

  with open('small_file.txt') as fp:

    print(count_vowels_v2(fp))

这一改动的主要变化在于提升了函数的适用面。因为 Python 是“鸭子类型”的,虽然函数需要接受文件对象,但实际上我们可以将任何实现了文件协议的“类文件对象(file-like object)”传入 count_vowels_v2 函数中。

Python 中存在许多“类文件对象”。例如 io 模块内的 StringIO 对象就是其中之一。它是一种基于内存的特殊对象,拥有与文件对象几乎一致的接口设计。

利用 StringIO,我们可以非常方便地为函数编写单元测试。


  # 注意:以下测试函数需要使用 pytest 执行

  import pytest

  from io import StringIO



  @pytest.mark.parametrize(

    "content,vowels_count", [

      # 使用 pytest 提供的参数化测试工具,定义测试参数列表

      # (文件内容, 期待结果)

      ('', 0),

      ('Hello World!', 3),

      ('HELLO WORLD!', 3),

      ('你好,世界', 0),

    ]

  )

  def test_count_vowels_v2(content, vowels_count):

    # 利用 StringIO 构造类文件对象 "file"

    file = StringIO(content)

    assert count_vowels_v2(file) == vowels_count

使用 pytest 运行测试可以发现,函数可以通过所有用例:


  ❯ pytest vowels_counter.py

  ====== test session starts ======

  collected 4 items


  vowels_counter.py ... [100%]


  ====== 4 passed in 0.06 seconds ======

让编写单元测试变得更简单,并非修改函数依赖后的唯一好处。除了 StringIO 外,subprocess 模块调用系统命令时用来存储标准输出的 PIPE 对象,也是一种“类文件对象”。这意味着我们可以直接将某个命令的输出传递给 count_vowels_v2 函数来计算元音字母数:


  import subprocess


  # 统计 /tmp 下面所有一级子文件名(目录名)有多少元音字母

  p = subprocess.Popen(['ls', '/tmp'], stdout=subprocess.PIPE, encoding='utf-8')


  # p.stdout 是一个流式类文件对象,可以直接传入函数

  # OUTPUT: 42

  print(count_vowels_v2(p.stdout))

正如之前所述,将函数参数修改为“文件对象”,最大的好处是提高了函数的适用面和可组合性。通过依赖更为抽象的“类文件对象”而非文件路径,为函数的使用方式开启了更多可能,StringIO、PIPE 以及任何其他满足协议的对象都可以成为函数的客户。

不过,这样的改造并非毫无缺点,它也会给调用方带来一些不便。假如调用方就是想要使用文件路径,那么就必须自行处理文件的打开操作。

如何编写兼容二者的函数

有没有办法既拥有“接受文件对象”的灵活性,又能让传递文件路径的调用方更方便?答案是:有,而且标准库中就有这样的例子。

打开标准库里的 xml.etree.ElementTree 模块,查看其中的 ElementTree.parse 方法。你会发现该方法既可以使用文件对象调用,也接受字符串的文件路径。而它实现这一点的手法也非常简单易懂:


  def parse(self, source, parser=None):

    """*source* is a file name or file object, *parser* is an optional parser

    """

    close_source = False

    # 通过判断 source 是否有 "read" 属性来判定它是不是“类文件对象”

    # 如果不是,那么调用 open 函数打开它并负担起在函数末尾关闭它的责任

    if not hasattr(source, "read"):

      source = open(source, "rb")

      close_source = True

使用这种基于“鸭子类型”的灵活检测方式,count_vowels_v2 函数同样可以被改造得更方便,此处不再赘述。

总结

总结

文件操作是日常工作中经常接触的领域,使用更方便的模块、利用生成器节约内存以及编写适用面更广的函数,可以让我们编写出更高效的代码。

让我们最后再总结一下吧:

  • 使用 pathlib 模块可以简化文件和目录相关的操作,并让代码更直观
  • PEP-519 定义了表示“文件路径”的标准协议,Path 对象实现了这个协议
  • 通过定义生成器函数来分块读取大文件可以节约内存
  • 使用 iter(callable,sentinel) 可以在一些特定场景简化代码
  • 难以编写测试的代码,通常也是需要改进的代码
  • 让函数依赖“类文件对象”可以提升函数的适用面和可组合性

以上就是本文的全部内容,希望对大家的学习有所帮助,也希望大家多多支持本站。

相关标签:

本站声明

本文内容仅供参考,不代表本站观点。转载或引用请注明出处。如发现内容问题,请通过联系我们页面反馈。

Python 相关产品和应用

全部下载

专题列表