Skip to content

Instantly share code, notes, and snippets.

@ptmcg
Last active May 5, 2022 00:44
Show Gist options
  • Star 0 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
  • Save ptmcg/0b70b2d033f33971f6af448672eb8dc3 to your computer and use it in GitHub Desktop.
Save ptmcg/0b70b2d033f33971f6af448672eb8dc3 to your computer and use it in GitHub Desktop.
#
# THE PROBLEM
#
# create a generator and list out its items
a = (n**2 for n in range(10))
print(list(a))
# listing it out again creates an empty list because the generator has been consumed;
# generators in this state will continue to raise StopIteration, but caller will take
# this to mean an empty generator
print(list(a))
#
# DEFINE A StrictGenerator WRAPPER
# that raises an exception if a consumed generator gets accessed again.
#
class AccessingConsumedGeneratorException(Exception):
"""
Exception raised if a generator is accessed after it has already raised StopIteration.
"""
class StrictGenerator:
"""
Wrapper for generator which will only permit it to be consumed once.
Additional accesses will raise AccessingConsumedGeneratorException.
"""
def __init__(self, gen):
self._gen = gen
self._gen_ended = False
def __iter__(self):
return self
def __next__(self):
try:
ret = next(self._gen)
return ret
except StopIteration:
if self._gen_ended:
ex = AccessingConsumedGeneratorException()
raise ex from None
self._gen_ended = True
raise
# wrap a generator in StrictGenerator
a = StrictGenerator(n**2 for n in range(10))
# consume a as usual
print(list(a))
# subsequent access of a will raise AccessingConsumedGeneratorException
print(list(a))
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment