In my discrete event simulation package ‘salabim’ (www.salabim.org), I use generators to model processes. These generators can yield several methods, like
yield self.hold(10)
If the user forgets the yield, like
self.hold(20)
, the program is syntactically still correct, but doesn’t work as expected and the cause is -particularly for less experienced modellers- rather difficult to understand.
Therefore, I would like to be able to check whether the method has indeed been yielded and not called directly.
I have made a check, which is based on checking the source code, which is not 100% foolproof, but sufficient for my purpose:
import inspect
import functools
class Component:
@functools.lru_cache
@staticmethod
def lines(co_code):
return inspect.getsource(co_code).split("\n")
def hold(self, x):
print("hold")
caller_frame = inspect.currentframe().f_back
co_code = caller_frame.f_code
rel = caller_frame.f_lineno - co_code.co_firstlineno
if not Component.lines(co_code)[rel].strip().startswith("yield "):
raise ValueError("not called as yield")
...
class X(Component):
def process(self):
yield self.hold(10)
self.hold(20)
x0 = X()
p0 = x0.process()
next(p0)
next(p0)
This demonstration code works but is neither very elegant nor very performant.
Is there a better way to check whether a method is indeed called with yield rather than directly?
Fair consideration! Unfortunately a tough one, but not impossible. There are a few possible approaches.
Firstly, static analysis as you have done. I would actually work slightly differently here, starting with an AST parse of the entire script, and doing some source-level analysis as a linter of sorts; that should be able to catch a lot of standard idioms and simple forgetfulness.
Secondly - and this is unreliable, but may serve you well - add a __del__ warning if the thing isn’t yielded. This is the same thing that’s done with async functions:
>>> async def spam(): pass
...
>>> def whoops():
... spam()
...
>>> whoops()
<stdin>:2: RuntimeWarning: coroutine 'spam' was never awaited
RuntimeWarning: Enable tracemalloc to get the object allocation traceback
>>>
There’s no guarantee that a __del__ method will be called (circular references and interpreter shutdown can lead to this), but it’ll catch a lot of simple “oops, forgot to put the yield keyword on it” situations (since those will immediately dispose of the object in question).
@Rosuav
Thanks for your response.
Unfortunately, static analysis is not possible as under certain circumstances the call should be done without yield. So my demonstration code is actually a simplification of the real problem.
I do 't really understand what you mean with a __del__ warning, Please explain.
What do you mean? Would the object be retained in some sort of collection and then yielded later?
Sure. Broadly speaking, what you want to do is know whether the thing has been yielded - which itself isn’t really a concept, but I’m assuming it gets passed along to something else that consumes it in some way. When the object is disposed of (in its __del__ method), it checks to see whether the data has been consumed. If it has, great! But if not, it assumes it must have been called and not yielded, and so it emits a warning.
Can you post a bit more of your code so we can see how this fits into the larger scheme of things?
No, if called (not yielded) it just performs some actions.
This still very much simplified code demonstrates it:
import inspect
import functools
class Component:
@functools.lru_cache
@staticmethod
def lines(co_code):
return inspect.getsource(co_code).split("\n")
def hold(self, x):
print("hold")
caller_frame = inspect.currentframe().f_back
co_code = caller_frame.f_code
rel = caller_frame.f_lineno - co_code.co_firstlineno
if Component.lines(co_code)[rel].strip().startswith("yield "):
if self is not current_component:
raise ValueError("yield is not allowed for non currect component")
else:
if self is current_component:
raise ValueError("current component can only be yielded")
...
class X(Component):
def process(self):
yield self.hold(2) # correct
x1.hold(3) # correct
self.hold(4) # error
yield x1.hold(5) # error
x0 = X()
x1 = X()
current_component = x0
p0 = x0.process()
next(p0)
next(p0)
If you want see how it works in practice, goto www.salabim.org. This is essentialy the same as the code above, only no checks (that I would like to introduce).
Ah. I don’t fully understand your API here, so it looks like my earlier suggestion may not have been applicable. I’ll have to delve deeper into the code before making any other recommendations; this is a bit of an unusual way to do things.