脚本如何自定义迭代对象状态

本文目录导读：

1. 基础迭代器实现
2. 使用生成器函数
3. 可重置迭代器
4. 并行迭代器（带状态同步）
5. 基于状态的迭代控制
6. 异步迭代器状态
7. 实际应用示例
8. 最佳实践

我来详细介绍如何自定义迭代对象的状态,包括基本方法和高级技巧。

基础迭代器实现

使用类实现迭代器

class RangeIterator:
    """自定义范围迭代器"""
    def __init__(self, start, end, step=1):
        self.current = start
        self.end = end
        self.step = step
        self._state = 'ready'  # 自定义状态
    def __iter__(self):
        self._state = 'iterating'
        return self
    def __next__(self):
        if self.current >= self.end:
            self._state = 'exhausted'
            raise StopIteration
        value = self.current
        self.current += self.step
        self._state = f'at_{value}'
        return value
    def get_state(self):
        """获取当前迭代器状态"""
        return {
            'state': self._state,
            'current': self.current,
            'end': self.end,
            'progress': (self.current - self.end) / self.end * 100 if self.end else 0
        }
# 使用示例
ri = RangeIterator(0, 10, 2)
for i in ri:
    print(f"值: {i}, 状态: {ri.get_state()}")

使用生成器函数

def stateful_generator(data):
    """带状态的生成器"""
    state = {'processed': 0, 'failed': 0, 'total': len(data)}
    for index, item in enumerate(data):
        try:
            # 处理数据
            result = item * 2
            state['processed'] += 1
            state['current_index'] = index
            yield result, state.copy()
        except Exception as e:
            state['failed'] += 1
            state['last_error'] = str(e)
            yield None, state.copy()
# 使用示例
data = [1, 2, 3, 'error', 5]
gen = stateful_generator(data)
for value, state in gen:
    print(f"值: {value}, 状态: {state}")

可重置迭代器

class ResettableIterator:
    """可重置状态的迭代器"""
    def __init__(self, data):
        self._original_data = data
        self.reset()
    def reset(self, new_data=None):
        """重置迭代器状态"""
        if new_data:
            self._original_data = new_data
        self.data = list(self._original_data)
        self.index = 0
        self.iterations = 0
        self.status = 'reset'
    def __iter__(self):
        self.iterations += 1
        self.status = f'iteration_{self.iterations}'
        return self
    def __next__(self):
        if self.index >= len(self.data):
            self.status = 'completed'
            raise StopIteration
        value = self.data[self.index]
        self.index += 1
        self.status = f'processing_{self.index}/{len(self.data)}'
        return value
    def peek(self, offset=0):
        """预览下一个元素而不改变状态"""
        peek_index = self.index + offset
        if peek_index < len(self.data):
            return self.data[peek_index]
        return None
# 使用示例
ri = ResettableIterator([1, 2, 3, 4, 5])
for item in ri:
    if item == 3:
        print(f"批处理状态: {ri.status}, 当前位置: {ri.index}")
# 重置迭代器
ri.reset()

并行迭代器（带状态同步）

class ParallelIterator:
    """并行迭代器，支持多个序列同步迭代"""
    def __init__(self, *iterables):
        self.iterables = iterables
        self.index = 0
        self.states = []
        # 初始化每个迭代器的状态
        for i, iterable in enumerate(iterables):
            if hasattr(iterable, '__iter__'):
                self.states.append({
                    'index': 0,
                    'iter': iter(iterable),
                    'exhausted': False
                })
    def __iter__(self):
        return self
    def __next__(self):
        result = []
        self.states = []
        for i, state in enumerate(self.states):
            if state['exhausted']:
                result.append(None)
            else:
                try:
                    value = next(state['iter'])
                    state['index'] += 1
                    result.append(value)
                except StopIteration:
                    state['exhausted'] = True
                    result.append(None)
        if all(s['exhausted'] for s in self.states):
            raise StopIteration
        return result, self.index
# 使用示例
a = [1, 2, 3]
b = ['a', 'b', 'c', 'd']
pi = ParallelIterator(a, b)
for values, index in pi:
    print(f"索引 {index}: {values}")

基于状态的迭代控制

class StateControlledIterator:
    """基于状态控制迭代行为"""
    def __init__(self, data, config=None):
        self.data = data
        self.config = config or {}
        self.state = {
            'index': 0,
            'skip_count': self.config.get('skip_threshold', 2),
            'error_count': 0,
            'last_operation': 'init'
        }
    def __iter__(self):
        return self
    def __next__(self):
        while self.state['index'] < len(self.data):
            value = self.data[self.state['index']]
            self.state['index'] += 1
            # 基于状态的条件跳过
            if self._should_skip(value):
                self.state['skip_count'] -= 1
                self.state['last_operation'] = 'skip'
                continue
            # 错误处理
            if self._is_error(value):
                self.state['error_count'] += 1
                self.state['last_operation'] = 'error'
                if self.state['error_count'] >= 3:
                    raise StopIteration("Too many errors")
                continue
            self.state['last_operation'] = 'return'
            return value
        raise StopIteration
    def _should_skip(self, value):
        """判断是否应该跳过"""
        if self.config.get('skip_values'):
            return value in self.config['skip_values']
        return False
    def _is_error(self, value):
        """判断是否为错误值"""
        return value < 0 if isinstance(value, (int, float)) else False
# 使用示例
data = [1, 2, -1, 3, 4, 5, -2, 6]
config = {'skip_values': [4], 'skip_threshold': 2}
sci = StateControlledIterator(data, config)
for value in sci:
    print(f"处理值: {value}, 状态: {sci.state}")

异步迭代器状态

import asyncio
class AsyncStateIterator:
    """异步迭代器状态管理"""
    def __init__(self, data):
        self.data = data
        self.state = {
            'total': len(data),
            'processed': 0,
            'status': 'created',
            'errors': []
        }
    def __aiter__(self):
        self.state['status'] = 'started'
        return self
    async def __anext__(self):
        if self.state['processed'] >= self.state['total']:
            self.state['status'] = 'completed'
            raise StopAsyncIteration
        # 模拟异步操作
        await asyncio.sleep(0.1)
        value = self.data[self.state['processed']]
        self.state['processed'] += 1
        self.state['status'] = f'processing_{self.state["processed"]}'
        return value, self.state.copy()
# 使用示例
async def main():
    data = [1, 2, 3, 4, 5]
    iterator = AsyncStateIterator(data)
    async for value, state in iterator:
        print(f"值: {value}, 状态: {state['status']}")
asyncio.run(main())

实际应用示例

class ChunkedFileIterator:
    """分块文件读取迭代器"""
    def __init__(self, filename, chunk_size=1024):
        self.filename = filename
        self.chunk_size = chunk_size
        self.state = {
            'total_size': 0,
            'read_size': 0,
            'chunk_count': 0,
            'done': False,
            'last_error': None
        }
    def __iter__(self):
        self.file = open(self.filename, 'rb')
        self.state['total_size'] = self._get_file_size()
        return self
    def __next__(self):
        if self.state['done']:
            self.file.close()
            raise StopIteration
        chunk = self.file.read(self.chunk_size)
        if not chunk:
            self.state['done'] = True
            self.state['status'] = 'completed'
            raise StopIteration
        self.state['read_size'] += len(chunk)
        self.state['chunk_count'] += 1
        self.state['progress'] = (self.state['read_size'] / 
                                 self.state['total_size'] * 100)
        return chunk, self.state.copy()
    def _get_file_size(self):
        import os
        return os.path.getsize(self.filename)
    def __del__(self):
        if hasattr(self, 'file') and not self.file.closed:
            self.file.close()
# 使用示例
iterator = ChunkedFileIterator('example.txt', chunk_size=512)
for chunk, state in iterator:
    print(f"读取块 {state['chunk_count']}: {state['progress']:.1f}%")

最佳实践

1. 状态持久化：使用 copy() 确保状态不变
2. 错误处理：在状态中记录错误信息
3. 性能优化：避免在每次迭代中创建新对象
4. 线程安全：使用锁保护共享状态

这些方法可以根据实际需求进行组合和扩展,创建灵活的状态管理迭代器。