stamp.py

"""
This is a simple context-manager that stores information about already written files on the file system.

The idea is to keep track of files that have already been written and that have not changed.
The benefit is, that we do not need to generate the code for things, that already exist and are up to date.

Because we do not know how the generated code would look like and what files are generated,
we can need to store such information in a separate file.

Basically we check for every generator, if it needs to be generated.
If so, we save it's name and the current Unix timestamp. After we are done with writing,
be dump that information into a simple file as JSON.

Properties maintained by the API:
* A set of key,value pairs, where each key can be a string and each value is a number (Unix timestamp)
* Serilaizing and Deserializing such information to JSON
* Loading/Saving information to the file system
* Call reset() in order to clean the fs cache.

IMPORTANT:
    This API only keeps track about files that have already been written.
    This NO cache.
"""
import json
import os
import stat
import sys
import time
from json import JSONDecodeError
from typing import Union, Optional


def timestamp() -> float:
    """ Return the current Unix timestamp """
    return time.time()


Number = Union[int, float]


class StampChecker:
    def __init__(self, stamp_file: str = ".stamps") -> None:
        """
        Create a new StampChecker context manager.
        :param stamp_file: The name if the stamp-file to load and write
        """
        self.stamp_file: str = stamp_file
        self.stamps: dict = {}  # Empty by default

        if self._exists(self.stamp_file) and self._isfile(self.stamp_file):
            # If the file exists and is a file, load it
            self.stamps = self._load_stamp()

    def __enter__(self) -> "StampChecker":
        """ Enables use of the with statement """
        return self

    def __exit__(self, *exc):
        """ Save the information about written files after we are done writing """
        self.save()
        return False

    #
    # Public API
    #

    def stamp(self, k: str, ts: Number = None) -> None:
        """
        Stamp a key with the systems current timestamp
        :param k     : Arbitrary string that serves as a unique identifier.
        :param ts    : Optional ts, that should be used instead of the current ts (default).

        """
        self.stamps[k] = ts or timestamp()

    def get_stamp(self, k: str) -> Optional[Number]:
        """ Return the timestamp of a given key or None """
        return self.stamps.get(k, None)

    def is_up_to_date(self, k: str, ts: Number) -> bool:
        """ Does the key is present and is it timestamp newer than :ts """
        stamp = self.get_stamp(k)
        return bool(stamp and stamp >= ts)

    def needs_write(self, k: str, ts: Number) -> bool:
        """ Checks if a key needs to be written and stamps it.
         This is equivalent to calling is_up_to_date and stamp individually.
        """
        if self.is_up_to_date(k, ts):
            return False
        self.stamp(k)
        return True

    def save(self) -> None:
        """ Save state to fs"""
        self._save_stamp()

    def reset(self) -> None:
        """ Clear all values"""
        self.stamps = {}

    @staticmethod
    def stat(path: str) -> os.stat_result:
        """ Get basic file system stats for a given file or directory. """
        return os.stat(path)

    #
    # Private API
    #
    def _load_stamp(self) -> dict:
        content: str = self._read(self.stamp_file)
        try:
            return json.loads(content)
        except JSONDecodeError:
            # Make the error clear
            print(f"{self.stamp_file} is invalid. Delete it or provide a valid path.", file=sys.stderr)
            return {}

    def _save_stamp(self) -> None:
        """ Save current state to file system """
        stamps: str = self._serialize()
        self._write(self.stamp_file, stamps)

    def _serialize(self) -> str:
        """ Transform current state to json """
        return json.dumps(self.stamps)

    @staticmethod
    def _write(file: str, content: str):
        with open(file, "w") as fd:
            fd.write(content)

    @staticmethod
    def _read(file: str) -> str:
        with open(file, "r") as fd:
            return fd.read()

    @staticmethod
    def _delete(file: str) -> None:
        os.remove(file)

    def _isfile(self, path: str) -> bool:
        try:
            st = self.stat(path)
        except OSError:
            return False
        return stat.S_ISREG(st.st_mode)

    def _exists(self, path: str) -> bool:
        try:
            self.stat(path)
        except FileNotFoundError:
            return False
        return True