tg-me.com/python_testit/1182
Last Update:
🧠 Хитрая Python-задача: «Генератор, который взламывает итераторы»
Уровень: 💥 продвинутый (Python 3.12+)
Представьте, что у вас есть ленивый генератор, перебирающий огромный набор данных. Нужно «клонировать» итератор — создать копию, продолжающую работу с того же места, хотя обычные генераторы так не умеют.
🎯 Цель
Напишите cloneable(generator)
, которая:
● Принимает любую функцию-генератор или итерируемый объект
● Возвращает объект с методом clone()
, создающим независимую копию текущего состояния
● Позволяет параллельно вызывать next()
у оригинала и всех клонов
gen = cloneable(range(100))
it1 = gen.clone()
it2 = gen.clone()
next(it1) # 0
next(it1) # 1
next(it2) # 0 ← независимая позиция
💡 Ограничения
● Нельзя полностью буферизовать всю последовательность
● Память O(k), где k — число активных клонов
● Только стандартная библиотека; асинхронность не требуется
🔍 Идея решения
● Храним значения в collections.deque по мере чтения из исходного итератора
● Для каждого клона ведём текущий индекс в буфере
● После сдвига всех клонов дальше минимальной позиции — удаляем хвост буфера
✅ Скелет реализации
from collections import deque
class CloneableIterator:
def __init__(self, iterable):
self._source = iter(iterable)
self._buffer = deque()
self._positions = []
def clone(self):
pos = 0
self._positions.append(pos)
idx = len(self._positions) - 1
def _iter():
nonlocal pos
while True:
if pos < len(self._buffer):
yield self._buffer[pos]
else:
try:
val = next(self._source)
except StopIteration:
return
self._buffer.append(val)
yield val
pos += 1
self._positions[idx] = pos
self._shrink()
return _iter()
def _shrink(self):
if not self._positions:
return
min_pos = min(self._positions)
for _ in range(min_pos):
self._buffer.popleft()
self._positions = [p - min_pos for p in self._positions]
def cloneable(iterable):
return CloneableIterator(iterable)
📌 Пример использования
gen = cloneable(i**2 for i in range(5))
a = gen.clone()
b = gen.clone()
print(next(a)) # 0
print(next(a)) # 1
print(next(b)) # 0
print(next(b)) # 1
print(next(b)) # 4
Попробуйте улучшить решение:
● Добавьте защиту от «зомби-клонов»
● Реализуйте __length_hint__()
● Или сделайте асинхронную версию acloneable() на базе async for
BY Python tests
Warning: Undefined variable $i in /var/www/tg-me/post.php on line 283
Share with your friend now:
tg-me.com/python_testit/1182