simple_file_cache.py

simple_file_cache.py

# -*- coding: utf-8 -*-
"""Store simple data between multiple executions of a script.

Python2.6+, no external dependencies.

A simple class to store small amounts data between
multiple executions of a script (for one-off scripts).

Copyright: Manuel Barkhau 2020 Dedicated to the public domain.
CC0: https://creativecommons.org/publicdomain/zero/1.0/

Usage:

with SimpleFileCache(name="mycache") as cache:
    assert isinstance(cache, SimpleFileCache)
    assert isinstance(cache.obj, dict)
    cache.obj['hello'] = u"wörld"

print(cache.path)
with open(cache.path) as fobj:
    print(fobj.read())

# later
with SimpleFileCache(name="mycache", mode="r") as cache:
    assert isinstance(cache.obj, dict)
    assert cache.obj['hello'] == u"wörld"
"""
from __future__ import unicode_literals

import os
import json
import time
import shutil
import tempfile
import functools
import hashlib
import inspect


class SimpleFileCache(object):

    def __init__(
        self,
        default_constructor=dict,
        dumps=functools.partial(json.dumps, indent=4, sort_keys=True),
        loads=json.loads,
        **kwargs
    ):
        mode = kwargs.get('mode', "w")

        if mode not in ("r", "w"):
            msg = "Invalid value mode='{0}'. Valid modes are: 'r' and 'w'".format(mode)
            raise ValueError(msg)

        self._mode = mode
        self._dumps = dumps
        self._loads = loads

        if 'cache_id' in kwargs:
            cache_id = kwargs['cache_id']
        else:
            caller_path = os.path.abspath((inspect.stack()[1])[1])
            cache_id = hashlib.sha1(caller_path.encode("utf-8")).hexdigest()

        name = kwargs.get('name', "default")
        tmp = tempfile.gettempdir()
        fname = cache_id + "_" + name
        self.path = os.path.join(tmp, "_simple_file_cache_" + fname)

        self._default_constructor = default_constructor

        if os.path.exists(self.path):
            with open(self.path, mode="rb") as fobj:
                self._in_data = fobj.read()

            try:
                self.obj = self._loads(self._in_data)
            except UnicodeError:
                self.obj = self._loads(self._in_data.decode("utf-8"))
        else:
            self._in_data = None
            self.obj = self._default_constructor()

    def __enter__(self):
        return self

    def __exit__(self, typ, val, tb):
        if typ or val or tb:
            return False

        if "w" not in self._mode:
            return

        out_data = self._dumps(self.obj)
        if not isinstance(out_data, bytes):
            out_data = out_data.encode("utf-8")

        if self._in_data == out_data:
            return

        nonce = str(time.time())
        tmp_file = self.path + nonce
        try:
            with open(tmp_file, mode="wb") as fobj:
                fobj.write(out_data)

            shutil.move(tmp_file, self.path)
        finally:
            if os.path.exists(tmp_file):
                os.unlink(tmp_file)


def main():
    with SimpleFileCache() as cache:
        assert isinstance(cache, SimpleFileCache)
        assert isinstance(cache.obj, dict)
        cache.obj['hello'] = u"wörld"

    print("cache_path:", cache.path)
    with open(cache.path) as fobj:
        print(fobj.read())

    # later
    with SimpleFileCache(mode="r") as cache:
        assert isinstance(cache.obj, dict)
        assert cache.obj['hello'] == u"wörld"


if __name__ == '__main__':
    main()