Last active
March 17, 2020 17:22
-
-
Save dotchetter/4713c40f3e48189c14879a8f77e549b8 to your computer and use it in GitHub Desktop.
Call functions through the PollCache object for only receiving output from the function if new output is detected.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
class PollCache: | |
""" | |
This object is designed to act as a cushion between a | |
function or other callable, and its caller, and only | |
give returns when new data from the function is identified. | |
This way you can pass functions to an instance of this class | |
and loop indefinitely, where only new results will be returned. | |
The PollCache object will treat every function and its | |
constellation of arguments and keyword arguments as unique, | |
meaning that you can use the same function or other callable | |
with different parameters in a loop, and it will not be | |
overwritten in the PollCache just because the function is | |
the same, but will be treated as its own cache. | |
This class uses the __call__ method as its main interface. | |
Call the instance of this class as you would a function. | |
The syntax is simply: | |
>> cache = PollCache() | |
>> cache(function_name, parameter1 = val, parameter2 = val2, ...) | |
:silent_first_call: | |
Boolean. Set it to True if the desired behavior is to | |
build up a cache, and only start returning values if | |
any of the cached method returns deviate. | |
Default is that all initial calls with a new function | |
or a past function with new arguments, will return | |
values. Suppress this with this parameter. | |
---- Example with default behavior-------------------------------------- | |
>> cache = PollCache() | |
>> cache(function, a = 1, b = 2) # Will produce a return value | |
>> cache(function, a = 1, b = 2) # Will NOT produce a return value (identical output) | |
>> cache(function, a = 10, b = 20) # Will produce a return value (new output) | |
----- Example with silent first call ----------------------------------- | |
>> cache = PollCache(silent_first_call = True) | |
>> cache(function, a = 1, b = 2) # Will NOT produce a return value | |
>> cache(function, a = 1, b = 2) # Will NOT produce a return value (identical output) | |
>> cache(function, a = 10, b = 20) # Will produce a return value (new output) | |
""" | |
def __init__(self, silent_first_call = False): | |
self.cached_polls = dict() | |
self.silent_first_call = silent_first_call | |
def __call__(self, func: 'function', *args, **kwargs): | |
try: | |
new_result = func(*args, **kwargs) | |
except: | |
raise | |
if not func in self.cached_polls.keys(): | |
self.cached_polls[func] = [ | |
{'args': args, | |
'kwargs': kwargs, | |
'result': new_result, | |
'calls': 1}] | |
return new_result if not self.silent_first_call else None | |
for call in self.cached_polls[func]: | |
if call['args'] == args and call['kwargs'] == kwargs and call['result'] != new_result: | |
call['result'] = new_result | |
call['calls'] += 1 | |
return new_result | |
if not [i for i in self.cached_polls[func] if i['args'] == args and i['kwargs'] == kwargs]: | |
self.cached_polls[func].append({ | |
'args': args, | |
'kwargs': kwargs, | |
'result': new_result, | |
'calls': 1}) | |
return new_result if not self.silent_first_call else None | |
if __name__ == '__main__': | |
""" | |
Testing the pollcache object can be done here. | |
This test has a simple function, print_numbers. | |
It'll just print out whatever you give it really. | |
There's an optional parameter: add_time, which will | |
artificially make the method return different valules | |
every minute. This will simulate a function that might | |
fetch data from the internet through an API or alike, | |
and the goal is to only hear from it if it returns | |
new data. | |
""" | |
from datetime import datetime | |
from pprint import pprint | |
from time import sleep | |
def print_numbers(*args, add_time = False): | |
if not add_time: | |
return args | |
return f'{args}, {datetime.now().minute}' | |
pollcache = PollCache(silent_first_call = True) | |
while True: | |
r = pollcache(print_numbers, 1,2,3, add_time = True) | |
if r: print(r) | |
sleep(1) | |
pprint(pollcache.cached_polls) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment