We often face an issue that our long running services need some parts to be reloadable without restart of such service. To achieve this I implemented simple abstract class that wraps any container we could use. You can find whole example at my Github page.
from abc import ABC, abstractmethod
import time
class Reloadable(ABC):
"""
This class wraps a container and expose all its methods.
It reloads its content after `reload_every_secs` from `path`.
"""
def __init__(self, path, reload_every_secs=60):
self._path = path
self._last_update = time.time()
self._reload_every_secs = reload_every_secs
self._initialize_data()
self.reload()
def reload(self):
self._last_update = time.time()
try:
with open(self._path) as f:
self._fill_data(f)
except FileNotFoundError:
self._initialize_data()
@abstractmethod
def _initialize_data(self):
"""Change container to initial state"""
pass
@abstractmethod
def _fill_data(self, fp):
"""Fill data from file to container self._data"""
pass
def access(self):
if time.time() - self._last_update > self._reload_every_secs:
self.reload()
def __getattr__(self, item):
return getattr(self._data, item)
def __len__(self):
self.access()
return self._data.__len__()
def __getitem__(self, item):
self.access()
return self._data.__getitem__(item)
def __setitem__(self, key, value):
self.access()
return self._data.__setitem__(key, value)
def __delitem__(self, key):
self.access()
return self._data.__delitem__(key)
def __iter__(self):
self.access()
return self._data.__iter__()
def __reversed__(self):
self.access()
return self._data.__reversed__()
def __contains__(self, item):
self.access()
return self._data.__contains__(item)
def __str__(self):
self.access()
return self._data.__str__()
As you can see this class just wraps not specified container and whenever we want to interact with it somehow we trigger function access(). To actually use it simply implement two methods _initialize_data and _fill_data and you will have your container reloaded every few second as you can specify.
class ReloadableList(Reloadable):
def _initialize_data(self):
self._data = []
def _fill_data(self, fp):
self._data = [line.strip() for line in fp if line.strip()]
Then we can use this just like regular list with the difference that it will update its content:
>>> from reloadable_containers import ReloadableList
>>> l = ReloadableList("content_list.txt", reload_every_secs=5)
>>> print(len(l))
7
Then you can add something to the file:
$ echo "bit.ly" > content_list.txt
And your running application will now respond differently:
>>> print(len(l))
8
>>> print(l[-1])
bit.ly
Similarly you can define json container:
import json
class ReloadableJson(Reloadable):
def _initialize_data(self):
self._data = {}
def _fill_data(self, fp):
self._data = json.load(fp)
And use it just like dictionary:
>>> from reloadable_containers import ReloadableList
>>> j = ReloadableJson("content.json")
>>> print(j.get("lastName", "MISSING"))
Smith