August 15, 2021By Jason Cronquist← Back to Blog

Experimental Config State in Python

I had previously written about a simple state-management object that I had created in Python. I thought it might be useful to add a little more context on an advanced use-case for the state-object.

In my home-automation project, I have a series of State Objects that cache the state defined in configuration files on disk. These configuration files can be updated at any time by any application, (vim, a web server, etc.). To reflect the state accurately on disk, I have created a Config File State Object that is invalidated whenever a change to the file on disk is detected.

""" State Object that is created from a config.  Invalidated when config changes on disk. """
from typing import (
    List,
    Union
)
from datetime import (
    datetime,
    timedelta
)

import src.utils.configs.file_utils as config
from .state_object import (StateObject, T)

DEFAULT_MIN_TIME_BETWEEN_DISK_CHECKS = timedelta(seconds=15)

class ConfigStateObject(StateObject):
    """ Monitors a specific path on disk and invalidates the object whenever a change to the file occurs. """
    __min_time_between_disk_test: timedelta
    __config_dependencies: List[str]

    __last_modified_time: int
    __last_checked: datetime
    __last_config_count: int
    __last_set: datetime

    def __init__(self, depends_on: Union[List[str], str], min_time_between_disk_test: timedelta = None):
        super().__init__(depends_on=None)
        self.__last_modified_time = datetime.utcnow()
        self.__last_checked = datetime.utcnow()
        self.__last_config_count = 0
        self.__last_set = datetime.utcnow()

        if min_time_between_disk_test is None:
            self.__min_time_between_disk_test = DEFAULT_MIN_TIME_BETWEEN_DISK_CHECKS
        else:
            self.__min_time_between_disk_test = min_time_between_disk_test

        if depends_on is None:
            self.__config_dependencies = []
        elif isinstance(depends_on, list):
            self.__config_dependencies = depends_on
        else:
            self.__config_dependencies = [depends_on]

    @property
    def valid(self) -> bool:
        """ returns: true if the config on disk has not changed. Otherwise false. """
        if not super().valid:
            return False

        if self.__last_set is None:
            return False

        now = datetime.utcnow()

        if now - self.__last_set > timedelta(days=1):
            self.invalidate()
            return False

        if now - self.__last_checked > self.__min_time_between_disk_test:
            self.__last_checked = now

            latest_config_count = sum([config.file_count(config_name) for config_name in self.__config_dependencies])
            if latest_config_count != self.__last_config_count:
                self.invalidate()
                return False

            latest_mod_time = max([config.last_mod_time(config_name) for config_name in self.__config_dependencies])
            if latest_mod_time != self.__last_modified_time:
                self.invalidate()
                return False
        return True

    def set(self, value: T) -> None:
        super().set(value)
        latest_mod_time = max([config.last_mod_time(config_name) for config_name in self.__config_dependencies])
        self.__last_modified_time = latest_mod_time
        self.__last_config_count = sum([config.file_count(config_name) for config_name in self.__config_dependencies])
        now = datetime.utcnow()
        self.__last_checked = now
        self.__last_set = now

Finally, you would use this state object whenever you want to automatically update the Python State that is tied to any config file on disk. An example of it's usage is below.

""" A Monitored Config's State """
import yaml

CONFIG_PATH_ON_DISK = "/path/to/config.yaml"

class Config:
    """ Wrapper for automatic reloading of a specific configuration file """

    def __init__():
        self.state_cache = ConfigStateObject(depends_on=CONFIG_PATH_ON_DISK)

    @property
    def state(self):
        """ Loads automations from Disk Configs """
        if not self.state_cache.valid:
            automations_conf = yaml.full_load(CONFIG_PATH_ON_DISK)
            self.state_cache.set(automations_conf)
        return self.state_cache.get()