Skip to content

Instantly share code, notes, and snippets.

@gvx
Forked from dotchetter/PollCache.py
Last active March 17, 2020 17:37
Show Gist options
  • Star 0 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
  • Save gvx/5af6b0696a98dc645ecf75828a1aeeea to your computer and use it in GitHub Desktop.
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.
# 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