遍历文件系统的类是否适合Python Iterator协议?兼论适用场景
我已经知道如何通过实现__next__和__iter__方法把类变成迭代器,现在想搞清楚:什么时候不该把类实现成迭代器? 尤其是当要遍历的集合满足以下条件时:
- 迭代器创建后,集合的内容仍未完全确定
- 集合不是由迭代器内部生成的
- 不按固定索引顺序遍历
就算类有类似"next"的方法,它还适合做迭代器吗?
我的核心场景是一个遍历文件系统的类:给定路径后,调用它的next方法能获取路径下(含子路径)的"下一个逻辑"目录,为了效率,它不会存储所有目录的完整集合。我现在实现的DirectoryIterator就是这类,代码如下:
import logging import os from pathlib import Path from typing import Union from repoManager.Models import ImagePromptDirectory from repoManager.utils import generate_file_name from utils.pathingUtils import get_reverse_sorted_directory_by_name, get_next_file_index_from_reverse_sorted, DIRECTION class DirectoryIterator: """ An iterator that simplifies traversal of the file system. Its primary mode of traversal is going to the logical "next" prompt directory in the system. When created the DirectoryIterator first makes a snapshot of the date directories and orders them by most recent. This snapshot can become stale so users of this Directory Iterator should make a new Directory Iterator to get a more recent snapshot of dates. Logically the repo directory structure can look as follows: ${pathToDirectories}/ 2024-01-11/ 15:05:06.713451_Sad rat/ 1.png 2.png 16:55:31.897695_Mom yelling/ 1.png Whatever the "pathToDirectories" location is there will be directories there named by iso-8601 date format. Within each directory a prompt, prefixed by the iso-8601 time used to generated it, exists. Within each timePrompt directory the images generated for that prompt are stored. When saved to the file system the actual directories won't be sorted, as they are actual stored by order of the file systems hash system. For optimization purposes that is why a "snapshot" of the dates is used to reduce having to sort all dates when loading them from the file system. The DirectoryIterator has internal pionters to where it is in the directory structure. These pointers start at the very first entry in the directory structure depending on the direction provided (default of Forward.) These pointers can be adjusted with a provided startingDirectory arg. The startingDirectory does not need to be physical directory in the system aand this DirectoryIterator will iterate to the next logical entry of the provided starting directory if that is the case. """ def __init__(self, pathToDirectories: Union[str, Path], startingDirectory: ImagePromptDirectory = None): self.pathToDirectories = Path(pathToDirectories) self.sortedDateDirectories = get_reverse_sorted_directory_by_name(self.pathToDirectories) if(startingDirectory is not None): # if directory provided (even if it doesn't exist) use that startingPromptWithTime = generate_file_name(startingDirectory.time, startingDirectory.prompt) self.currentDate = startingDirectory.date self.currentTimePromptDirectories = get_reverse_sorted_directory_by_name(self.pathToDirectories/self.currentDate) self.currentTimePrompt = startingPromptWithTime else: self.currentDate = None self.currentTimePromptDirectories = [] self.currentTimePrompt = None def _rewind_to_right(self): self.currentDate = "9999-99-99" # I would be blessed if I lived to see this be an error self.currentTimePromptDirectories = [] self.currentTimePrompt = "" def _rewind_to_left(self): self.currentDate = "0000-00-00" self.currentTimePromptDirectories = [] self.currentTimePrompt = "" def _rewind_if_necessary(self, direction: DIRECTION = DIRECTION.FORWARD): if(self.get_current_image_prompt_directory() is None): if(direction is DIRECTION.FORWARD): self._rewind_to_right() else: self._rewind_to_left() def get_current_image_prompt_directory(self) -> ImagePromptDirectory: """ Gets the current image prompt that this iterator is pointing to. Returns ------- ImagePromptDirectory A directory model that could represent a physical entry. However the physical entry could of been deleted externally since the last time this iterator took a snapshot of the time_prompts directories for a given date. """ logging.debug(f'Getting the current directory') if( self.currentTimePrompt is not None and self.pathToDirectories is not None and self.currentDate is not None ): logging.error(f'self.currentTimePrompt {self.currentTimePrompt}') time, prompt = self.currentTimePrompt.split("_") return ImagePromptDirectory( prompt=prompt, time=time, # Get only the repo name. Not the full absolute path of the repo. repo=os.path.basename(os.path.normpath(self.pathToDirectories)), date=self.currentDate, ) return None def _iterate_time_prompt(self, direction: DIRECTION = DIRECTION.FORWARD) -> ImagePromptDirectory: # attempt to get next time prompt within current date directory nextTimePromptIndex = get_next_file_index_from_reverse_sorted(fileName=self.currentTimePrompt, reverseSortedFiles=self.currentTimePromptDirectories, direction=direction) if(nextTimePromptIndex is not None): nextTimePrompt = self.currentTimePromptDirectories[nextTimePromptIndex] self.currentTimePrompt = nextTimePrompt return self.get_current_image_prompt_directory() self.currentTimePrompt = None return None def _iterate_date(self, direction: DIRECTION = DIRECTION.FORWARD) -> ImagePromptDirectory: # attempt to get next time prompt within current date directory nextDateIndex = get_next_file_index_from_reverse_sorted(fileName=self.currentDate, reverseSortedFiles=self.sortedDateDirectories, direction=direction) if(nextDateIndex is not None): nextDate = self.sortedDateDirectories[nextDateIndex] nextTimePromptDirectories = get_reverse_sorted_directory_by_name(self.pathToDirectories/nextDate) startIndexOfNextDateFolder = 0 if direction is not DIRECTION.BACKWARD else len(nextTimePromptDirectories)-1 nextTimePrompt = nextTimePromptDirectories[startIndexOfNextDateFolder] self.currentDate = nextDate self.currentTimePromptDirectories = nextTimePromptDirectories self.currentTimePrompt = nextTimePrompt return self.get_current_image_prompt_directory() self.currentDate = None self.currentTimePromptDirectories = None self.currentTimePrompt = None return None def get_next_time_prompt_directories(self, direction: DIRECTION = DIRECTION.FORWARD) -> ImagePromptDirectory: """ Gets the logical next image prompt that this iterator from the current image prompt this iterator is pointing to. It could also return None if all image prompts have been exhuasted. Not garanteed to be 100% accurate in scenarios where the physical file system has been externally modified. Parameters ---------- direction: (DIRECTION): The direciton to get the next prompt time directory. Supports "forward" or "backward". Default is to go forward. Returns ------- ImagePromptDirectory A directory model that could represent the next logical physical entry. However the physical entry could of been deleted externally since the last time this iterator took a snapshot of the time_prompts directories for a given date. """ logging.debug(f'Getting next prompt with direction {direction}') self._rewind_if_necessary(direction=direction) # get next time prompt if current date direcotry has any candidate_time_prompt_directory = self._iterate_time_prompt(direction) # Iterate through date directories until a time prompt directory is found or all date directories are exhuasted while(self.currentDate is not None and candidate_time_prompt_directory is None): candidate_time_prompt_directory = self._iterate_date(direction) return candidate_time_prompt_directory
这个类的核心"next"逻辑由get_next_time_prompt_directories方法实现。只需要做少量修改(比如把direction参数移到构造方法、遍历完毕时抛出StopIteration、替换成__iter__和__next__方法),就能转成标准Python迭代器。但它靠快照子目录提升效率的行为,让我怀疑这么做是否符合Python风格,或许它更适合叫"scanner"?
先明确迭代器的核心约定
Python迭代器的核心是一次性、单向、状态绑定:
- 迭代器是有状态的,一旦开始遍历,就沿着固定方向推进,不能随意跳步或反向(除非自定义实现,但不符合常规预期)
- 迭代器遍历完就耗尽了,不能重复使用(除非重新实例化)
- 用户预期迭代器返回的序列是确定的、稳定的(至少在遍历期间是这样)
什么时候不该用迭代器?
你的三个条件刚好戳中迭代器的痛点:
- 集合内容不确定:迭代器创建后,如果集合还在变化(比如文件系统被外部修改),迭代器返回的结果会和实际情况脱节,违反用户对"遍历完整序列"的预期。
- 集合不是内部生成:迭代器通常要么遍历自己生成的序列,要么绑定一个固定的外部序列。如果集合是外部动态的(比如文件系统),迭代器的快照机制会导致数据过时,这时候用户更需要的是"实时扫描"而非"固定遍历"。
- 不按固定索引遍历:迭代器的
__next__默认是单向推进的,如果你需要随时切换方向(比如你的direction参数),强行用迭代器会让接口变得别扭——用户预期迭代器用for循环就能从头到尾走一遍,而不是还要每次调用时指定方向。
回到你的DirectoryIterator
你的类更适合叫Scanner(扫描器),原因很简单:
- 它的核心是动态扫描外部资源,而不是遍历一个预先确定的序列
- 它支持双向切换方向,这完全不符合迭代器"单向推进"的常规用法
- 快照机制意味着它的结果不是实时的,用户需要明确知道什么时候需要重新扫描,而迭代器用户通常不会有这个心理预期
如果硬改成迭代器,会有这些问题:
- 你得把
direction移到构造方法,这样要反向遍历就得重新实例化一个迭代器,很繁琐 - 用户用
for dir in your_iterator:的时候,会默认遍历的是当前文件系统的完整状态,但实际上是快照,容易踩坑 - 迭代器耗尽后就没用了,而你的扫描器其实可以重置状态继续用,这会浪费它的灵活性
那该怎么设计?
保持现在的get_next_time_prompt_directories方法,把它做成一个扫描器类,同时可以额外实现一个迭代器适配器——当用户需要用迭代器风格遍历的时候,提供一个生成器方法,比如:
def iterate(self, direction: DIRECTION = DIRECTION.FORWARD): while True: item = self.get_next_time_prompt_directories(direction) if item is None: break yield item
这样用户既可以灵活调用get_next方法做动态扫描,也可以用for dir in scanner.iterate():做一次性遍历,兼顾了灵活性和Pythonic的用法。
总结:迭代器适合遍历固定、单向、一次性的序列;而扫描器适合动态、可切换方向、需要实时感知外部变化的场景。你的类明显属于后者,叫Scanner更合适,也更符合用户的预期。
内容的提问来源于stack exchange,提问作者Cory Bond

