Flag to test if the generator is "running" and / or is already exhausted
Summary
What I need is a way to determine if the generator is currently running.
More details
My definition is "currently working":
- Started executing generator code (i.e.
next
or.send(None)
or some other iteration operation such as a for loop had previously been performed on the generator, AND - the generator has not previously been exhausted / closed (again note that for the above described generator the generator is
.throw(some_error)
caught and will not exhaust the generator, since some new cycle is started, no cycle oryield from
the error being handled will usually be generator exhaust).
I understand that there is no way (or at least trivially simple) to "look" ahead into the generator and see if it will pick up StopIteration
on the next iteration. This is contrary to the nature of a generator and is NOT who I am.
Example
Let's say I have a generator g
created by some generator function f
:
def f():
'''Lots of 1s.'''
print('I am now running.')
while True:
try:
signal = yield 1
if signal:
break
except GeneratorExit:
raise # important so that closing the generator doesn't cause a RuntimeError
except Exception:
print('caught exception!')
We can "initialize" the generator like this:
>>> g = f()
>>> next(g)
I am now running.
1
# OR:
>>> g.send(None) # equivalent to next(g)
I am now running.
1
If I send anything truthful to the generator, the error StopIteration
will be raised (for example g.send('foo')
). Alternatively, if I call g.close()
and then do next(g)
, I also get StopIteration
. If I clear any errors with g.throw(e)
, the generator continues. This is all as expected.
What I would like to do is something like the following, which shows if this generator is working:
>>> g=f()
>>> g.running
False
>>> g.send(None)
I am now running.
1
>>> g.running
True
>>> g.close()
>>> g.running
False
In the answer, I gave one possible approach. However, I think there must be a better way.
source to share
inspect.getgeneratorstate
reports the state of your generator:
>>> import inspect
...
... def gen():
... yield 1
...
... g = gen()
...
>>> inspect.getgeneratorstate(g)
'GEN_CREATED'
>>> next(g)
1
>>> inspect.getgeneratorstate(g)
'GEN_SUSPENDED'
>>> next(g)
>>> inspect.getgeneratorstate(g)
'GEN_CLOSED'
source to share
My only idea is some kind of generator wrapper with methods that delegate to a stored generator, but this is a bit cumbersome. It might look something like this (haven't tested this):
class MyGen():
'''Generator wrapper with 'exhausted' and 'running' flags.'''
def __new__(cls, some_func):
cls._func = some_func
return super().__new__(cls)
def __call__(self, *args, **kwargs):
self.exhausted = False
self.running = False
self.start(*args, **kwargs)
return self
def start(self, *args, **kwargs):
self._gen = self._func(*args, **kwargs)
def __next__(self):
try:
self.running = True
next(self._gen)
except StopIteration:
self.exhausted = True
self.running = False
raise
def send(self, *args, **kwargs):
try:
self.running = True
self._gen.send(*args, **kwargs)
except StopIteration:
self.exhausted = True
self.running = False
raise
def throw(self, *args, **kwargs):
try:
self._gen.throw(*args, **kwargs)
except StopIteration:
self.exhausted = True
self.running = False
raise
def close(self):
self._gen.close()
self.exhausted = True
self.running = False
You can even use this as a decorator:
@MyGen
def f():
yield 1
I really don't like this because it looks like there must actually be an existing place to look for the "generator exhausted" or "generator running" flag.
source to share