-
-
Save gvx/5af6b0696a98dc645ecf75828a1aeeea 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
# disclaimer: untested code ahead! see https://www.reddit.com/r/Python/comments/fk4wal/i_needed_an_object_that_can_with_one_instance/fkrbodb/ | |
from dataclasses import dataclass | |
from enum import Enum, auto | |
from typing import Any, Callable, Tuple | |
@dataclass | |
class CacheValue: | |
result: Any | |
times_changed: int = 1 | |
class CacheChanged(Enum): | |
FIRST_CALL = auto() | |
CHANGED = auto() | |
SAME = auto() | |
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, *args, **kwargs) -> (changed, value) | |
where value = function(*args, **kwargs) | |
and changed is CacheChanged.FIRST_CALL if it was the first call to function(*args, **kwargs) | |
CacheChanged.SAME if the value is the same as the last time that cache was called with the same arguments | |
CacheChanged.CHANGED otherwise | |
Note that the values in args and kwargs need to be hashable. | |
""" | |
def __init__(self) -> None: | |
self.cached_polls = {} | |
def __call__(self, func: Callable, *args: Any, **kwargs: Any) -> Tuple[CacheChanged, Any]: | |
new_value = func(*args, **kwargs) | |
cache_key = (func, args, tuple(kwargs.items())) | |
try: | |
cache_value = self.cached_polls[cache_key] | |
except KeyError: | |
self.cached_polls[cache_key] = CacheValue(new_value) | |
return CacheChanged.FIRST_CALL, new_value | |
if cache_value.result != new_value: | |
cache_value.result = new_value | |
cache_value.times_changed += 1 | |
return CacheChanged.CHANGED, new_value | |
return CacheChanged.SAME, new_value | |
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() | |
try: | |
while True: | |
changed, r = pollcache(print_numbers, 1, 2, 3, add_time=True) | |
if changed is CacheChanged.CHANGED: print(r) | |
sleep(1) | |
except KeyboardInterrupt: | |
pprint(pollcache.cached_polls) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment