In this blog post I want to just share a little Python trick that I'm about to go and use in my aiohivebot project.
I was strugling with the problem of context in the user defined aiohivebot methods.
Let's say we have the following user defined method in our bot:
async def vote_operation(self, body):
"""Handler for cote_operation type operations in the HIVE block stream"""
if "voter" in body and "author" in body and "permlink" in body:
result = await self.bridge.get_post(author=body["author"], permlink=body["permlink"])
content = result.result()
if content and "is_paidout" in content and content["is_paidout"]:
print("Vote by", body["voter"], "on expired post detected: @" + \
body["author"] + "/" + body["permlink"] )
This is fine for most purposes, but what if we want block level information, for example the timestamp from the block? Maybe we want the transaction id from transaction level. If instead of an operation we had a method for a HiveEngine contract operation, maybe we would want context from any number of contexts at different levels.
We could just say that the user defined methods should always have all of these contexts in their function signature, but that wouldn't be convenient and linters will complain about unused arguments.
async def vote_operation(self, body, operation_context, transaction_context, block_context):
...
This is far from ideal to mandate this. What we want, idealy, is the posibility to just define the available arguments that we want to use. So in this example, if we want the timestamp from the block context, we simply define:
async def vote_operation(self, body, block_context):
...
And it should just work.
But how do we make it fit.
Right now, the code in aiohivebot for invoking this user defined method looks something like this:
# If the derived class has a "operation" callback, invoke it
if hasattr(self, "operation"):
await getattr(self, "operation")(operation=operation,
tid=transaction_ids[index],
transaction=transactions[index],
block=block)
# If the derived class has an operation type specificcallback, invoke it
if "type" in operation and "value" in operation and hasattr(self, operation["type"]):
await getattr(self, operation["type"])(body=operation["value"])
Way too static. What we are going to do is change things to that the above code will look as follows but without breaking anything:
# If the derived class has a "operation" callback, invoke it
await self.forwarder.operation(operation=operation,
tid=transaction_ids[index],
transaction=transactions[index],
block=block)
# If the derived class has an operation type specificcallback, invoke it
if "type" in operation and "value" in operation:
if hasattr(self, operation["type"]):
await getattr(self.forwarder, operation["type"](
body=operation["value"],
operation=operation,
tid=transaction_ids[index],
transaction=transactions[index],
block=block
)
See how the per operation type method now gets called with many more arguments? The underlying method however still gets called with only the body argument.
So how do we do this?
Well, first we add a wrapper to the constructor ot our BaseBot class:
def __init__(self, start_block=None, roll_back=0, roll_back_units="blocks"):
...
self.forwarder = ObjectForwarder(self)
...
Now comes the fun part, defining this wrapper:
class ObjectForwarder:
def __init__(self, bot):
self.bot = bot
def __getattr__(self, methodname):
if hasattr(self.bot, methodname):
return FunctionForwarder(getattr(self.bot, methodname))
return FunctionNull()
The getattr method gets called when an attribute or method gets accessed that isn't explicitly defined. This allows for pretty fun logic in Python code.
This wrapper class now has every conceivable method. One implemented by FunctionNull for methods not implemented by the bot.
This is a do nothing method:
class FunctionNull:
async def __call__(self, *args, **kwargs):
pass
And for each method defined by the user code, ther is a wrapper methos defined by the FunctionForwarder class
class FunctionForwarder:
def __init__(self, method):
self.method = method
self.args = set(inspect.signature(method).parameters.keys())
async def __call__(self, *args, **kwargs):
droplist = []
for key in kwargs:
if key not in self.args:
droplist.append(key)
for key in droplist:
kwargs.pop(key)
unknown = self.args - set(kwargs.keys())
if unknown:
raise ValueError("Non standard named arguments for method:" + str(list(unknown)))
self.method(**kwargs)
This is really the interesting part of our Python trick. In our constructor we use the inspect module to get the list od method arguments that the user code defines. We hold on to this info in a set we will end up using when the special call method gets invoked.
We see that call geta a special argument **kwargs , that contains a dict of all the keyword arguments the dynamic method was invoked with.
Now what we do is that we walk through the keys of this dict and every argument that isn't expected by the the actual method gets errazed prior to invocation.
We do a quick check to see all user defined arguments are actually valid, and then we invoke the actual method.
I think these three iny classes using __getattr__, __call__ and inspect, combine into a quite usefull pattern when combined with optional user definable methods with optional arguments that can be made part of the actual function fingerprint at the users discretion. Basicly all unused arguments ge erased in invocation avoiding linters complaints about unused function arguments, and avoiding user code that looks more complicated.
Congratulations @pibara! You have completed the following achievement on the Hive blockchain And have been rewarded with New badge(s)
Your next target is to reach 600 posts.
You can view your badges on your board and compare yourself to others in the Ranking
If you no longer want to receive notifications, reply to this comment with the word
STOPTo support your work, I also upvoted your post!
Check out our last posts: