DDNS/ddns/util/cache.py

# -*- coding: utf-8 -*-
"""
cache module
文件缓存
"""


from os import path, stat
from pickle import dump, load
from time import time
from logging import getLogger, Logger  # noqa: F401


class Cache(dict):
    """
    using file to Cache data as dictionary
    """

    def __init__(self, path, logger=None, sync=False):
        # type: (str, Logger | None, bool) -> None
        self.__data = {}
        self.__filename = path
        self.__sync = sync
        self.__time = time()
        self.__changed = False
        self.__logger = (logger or getLogger()).getChild("Cache")
        self.load()

    @property
    def time(self):
        """
        缓存修改时间
        """
        return self.__time

    def load(self, file=None):
        """
        load data from path
        """
        if not file:
            file = self.__filename

        self.__logger.debug("load cache data from %s", file)
        if path.isfile(file):
            with open(self.__filename, "rb") as data:
                try:
                    self.__data = load(data)
                    self.__time = stat(file).st_mtime
                    return self
                except ValueError:
                    pass
                except Exception as e:
                    self.__logger.warning(e)
        else:
            self.__logger.info("cache file not exist")

        self.__data = {}
        self.__time = time()
        self.__changed = True
        return self

    def data(self, key=None, default=None):
        # type: (str | None, Any | None) -> dict | Any
        """
        获取当前字典或者制定得键值
        """
        if self.__sync:
            self.load()

        if key is None:
            return self.__data
        else:
            return self.__data.get(key, default)

    def sync(self):
        """Sync the write buffer with the cache files and clear the buffer."""
        if self.__changed:
            with open(self.__filename, "wb") as data:
                dump(self.__data, data)
                self.__logger.debug("save cache data to %s", self.__filename)
            self.__time = time()
            self.__changed = False
        return self

    def close(self):
        """Sync the write buffer, then close the cache.
        If a closed :class:`FileCache` object's methods are called, a
        :exc:`ValueError` will be raised.
        """
        self.sync()
        del self.__data
        del self.__filename
        del self.__time
        self.__sync = False

    def __update(self):
        self.__changed = True
        if self.__sync:
            self.sync()
        else:
            self.__time = time()

    def clear(self):
        if self.data() is not None:
            self.__data = {}
            self.__update()

    def __setitem__(self, key, value):
        if self.data(key) != value:
            self.__data[key] = value
            self.__update()

    def __delitem__(self, key):
        if key in self.data():
            del self.__data[key]
            self.__update()

    def __getitem__(self, key):
        return self.data(key)

    def __iter__(self):
        for key in self.data():
            yield key

    def __len__(self):
        return len(self.data())

    def __contains__(self, key):
        return key in self.data()

    def __str__(self):
        return self.data().__str__()

    def __del__(self):
        self.close()