X-Git-Url: https://wannabe.guru.org/gitweb/?a=blobdiff_plain;f=persistent.py;h=b42a5c0e3036a35d76dc0c8f32374d029e94fd53;hb=f995224a77488f3469c16a3f504dcf7a1393d834;hp=d62dd6754eeffc78c1c09adea7c82e778f8450b3;hpb=e6f32fdd9b373dfcd100c7accb41f57d83c2f0a1;p=python_utils.git

diff --git a/persistent.py b/persistent.py
index d62dd67..b42a5c0 100644
--- a/persistent.py
+++ b/persistent.py
@@ -1,11 +1,17 @@
 #!/usr/bin/env python3
 
-from abc import ABC, abstractmethod
+# © Copyright 2021-2022, Scott Gasch
+
+"""A Persistent is just a class with a load and save method.  This
+module defines the Persistent base and a decorator that can be used to
+create a persistent singleton that autoloads and autosaves."""
+
 import atexit
 import datetime
 import enum
 import functools
 import logging
+from abc import ABC, abstractmethod
 from typing import Any
 
 import file_utils
@@ -58,12 +64,29 @@ class Persistent(ABC):
 
 
 def was_file_written_today(filename: str) -> bool:
-    """Returns True if filename was written today."""
+    """Returns True if filename was written today.
+
+    >>> import os
+    >>> filename = f'/tmp/testing_persistent_py_{os.getpid()}'
+    >>> os.system(f'touch {filename}')
+    0
+    >>> was_file_written_today(filename)
+    True
+    >>> os.system(f'touch -d 1974-04-15T01:02:03.99 {filename}')
+    0
+    >>> was_file_written_today(filename)
+    False
+    >>> os.system(f'/bin/rm -f {filename}')
+    0
+    >>> was_file_written_today(filename)
+    False
+    """
 
     if not file_utils.does_file_exist(filename):
         return False
 
     mtime = file_utils.get_file_mtime_as_datetime(filename)
+    assert mtime is not None
     now = datetime.datetime.now()
     return mtime.month == now.month and mtime.day == now.day and mtime.year == now.year
 
@@ -75,11 +98,27 @@ def was_file_written_within_n_seconds(
     """Returns True if filename was written within the pas limit_seconds
     seconds.
 
+    >>> import os
+    >>> filename = f'/tmp/testing_persistent_py_{os.getpid()}'
+    >>> os.system(f'touch {filename}')
+    0
+    >>> was_file_written_within_n_seconds(filename, 60)
+    True
+    >>> import time
+    >>> time.sleep(2.0)
+    >>> was_file_written_within_n_seconds(filename, 2)
+    False
+    >>> os.system(f'/bin/rm -f {filename}')
+    0
+    >>> was_file_written_within_n_seconds(filename, 60)
+    False
     """
+
     if not file_utils.does_file_exist(filename):
         return False
 
     mtime = file_utils.get_file_mtime_as_datetime(filename)
+    assert mtime is not None
     now = datetime.datetime.now()
     return (now - mtime).total_seconds() <= limit_seconds
 
@@ -126,7 +165,7 @@ class persistent_autoloaded_singleton(object):
         self.instance = None
 
     def __call__(self, cls: Persistent):
-        @functools.wraps(cls)
+        @functools.wraps(cls)  # type: ignore
         def _load(*args, **kwargs):
 
             # If class has already been loaded, act like a singleton
@@ -134,35 +173,36 @@ class persistent_autoloaded_singleton(object):
             # memory.
             if self.instance is not None:
                 logger.debug(
-                    f'Returning already instantiated singleton instance of {cls.__name__}.'
+                    'Returning already instantiated singleton instance of %s.', cls.__name__
                 )
                 return self.instance
 
             # Otherwise, try to load it from persisted state.
             was_loaded = False
-            logger.debug(f'Attempting to load {cls.__name__} from persisted state.')
+            logger.debug('Attempting to load %s from persisted state.', cls.__name__)
             self.instance = cls.load()
             if not self.instance:
                 msg = 'Loading from cache failed.'
                 logger.warning(msg)
-                logger.debug(f'Attempting to instantiate {cls.__name__} directly.')
+                logger.debug('Attempting to instantiate %s directly.', cls.__name__)
                 self.instance = cls(*args, **kwargs)
             else:
-                logger.debug(
-                    f'Class {cls.__name__} was loaded from persisted state successfully.'
-                )
+                logger.debug('Class %s was loaded from persisted state successfully.', cls.__name__)
                 was_loaded = True
 
             assert self.instance is not None
 
             if self.persist_at_shutdown is PersistAtShutdown.ALWAYS or (
-                not was_loaded
-                and self.persist_at_shutdown is PersistAtShutdown.IF_NOT_LOADED
+                not was_loaded and self.persist_at_shutdown is PersistAtShutdown.IF_NOT_LOADED
             ):
-                logger.debug(
-                    'Scheduling a deferred called to save at process shutdown time.'
-                )
+                logger.debug('Scheduling a deferred called to save at process shutdown time.')
                 atexit.register(self.instance.save)
             return self.instance
 
         return _load
+
+
+if __name__ == '__main__':
+    import doctest
+
+    doctest.testmod()