From: Scott Gasch <scott@gasch.org>
Date: Wed, 22 Sep 2021 01:25:50 +0000 (-0700)
Subject: Persistent state.
X-Git-Url: https://wannabe.guru.org/gitweb/?a=commitdiff_plain;h=c46018e2b1ddc78f9df557c3fb24d2c2c849f054;p=python_utils.git

Persistent state.
---

diff --git a/persistent.py b/persistent.py
new file mode 100644
index 0000000..30e4ccb
--- /dev/null
+++ b/persistent.py
@@ -0,0 +1,89 @@
+#!/usr/bin/env python3
+
+from abc import ABC, abstractmethod
+import atexit
+import functools
+import logging
+
+import dill
+
+import file_utils
+
+
+logger = logging.getLogger(__name__)
+
+
+class Persistent(ABC):
+    @abstractmethod
+    def save(self):
+        pass
+
+    @abstractmethod
+    def load(self):
+        pass
+
+
+class persistent_autoload_singleton(Persistent):
+    def __init__(self, filename: str, *, max_age_sec: int = 0):
+        self.filename = filename
+        self.max_age_sec = max_age_sec
+        self.instance = None
+
+    def __call__(self, cls):
+        @functools.wraps(cls)
+        def _load(*args, **kwargs):
+
+            # If class has already been loaded, act like a singleton
+            # and return a reference to the one and only instance in
+            # memory.
+            if self.instance is not None:
+                logger.debug(
+                    f'Returning already instantiated singleton instance of {cls.__name__}.'
+                )
+                return self.instance
+
+            if not self.load():
+                assert self.instance is None
+                logger.debug(
+                    f'Instantiating {cls.__name__} directly.'
+                )
+                self.instance = cls(*args, **kwargs)
+
+            # On program exit, save state to disk.
+            atexit.register(self.save)
+            assert self.instance is not None
+            return self.instance
+        return _load
+
+    def load(self) -> bool:
+        if (
+                file_utils.does_file_exist(self.filename)
+                and (
+                    self.max_age_sec == 0 or
+                    file_utils.get_file_mtime_age_seconds(self.filename) <= self.max_age_sec
+                )
+        ):
+            logger.debug(
+                f'Attempting to load from file {self.filename}'
+            )
+            try:
+                with open(self.filename, 'rb') as f:
+                    self.instance = dill.load(f)
+                    return True
+            except Exception:
+                self.instance = None
+                return False
+        return False
+
+    def save(self) -> bool:
+        if self.instance is not None:
+            logger.debug(
+                f'Attempting to save {type(self.instance).__name__} to file {self.filename}'
+            )
+            try:
+                with open(self.filename, 'wb') as f:
+                    dill.dump(self.instance, f, dill.HIGHEST_PROTOCOL)
+                return True
+            except Exception:
+                return False
+        return False
diff --git a/tests/decorator_utils_test.py b/tests/decorator_utils_test.py
new file mode 100755
index 0000000..195dd63
--- /dev/null
+++ b/tests/decorator_utils_test.py
@@ -0,0 +1,24 @@
+#!/usr/bin/env python3
+
+import unittest
+
+import decorator_utils as du
+
+import unittest_utils as uu
+
+
+class TestDecorators(unittest.TestCase):
+
+    def test_singleton(self):
+
+        @du.singleton
+        class FooBar():
+            pass
+
+        x = FooBar()
+        y = FooBar()
+        self.assertTrue(x is y)
+
+
+if __name__ == '__main__':
+    unittest.main()