Reduce the doctest lease duration...
[python_utils.git] / persistent.py
index 8832572e7b032ae71a7898ca6b21743cf7ce45bf..58014608f9a894fa40742249cd9c04933007d52e 100644 (file)
@@ -1,5 +1,11 @@
 #!/usr/bin/env python3
 
+# © Copyright 2021-2022, Scott Gasch
+
+"""A :class:`Persistent` is just a class with a load and save method.  This
+module defines the :class:`Persistent` base and a decorator that can be used to
+create a persistent singleton that autoloads and autosaves."""
+
 import atexit
 import datetime
 import enum
@@ -16,30 +22,27 @@ logger = logging.getLogger(__name__)
 class Persistent(ABC):
     """
     A base class of an object with a load/save method.  Classes that are
-    decorated with @persistent_autoloaded_singleton should subclass this
-    and implement their save() and load() methods.
-
+    decorated with :code:`@persistent_autoloaded_singleton` should subclass
+    this and implement their :meth:`save` and :meth:`load` methods.
     """
 
     @abstractmethod
     def save(self) -> bool:
         """
         Save this thing somewhere that you'll remember when someone calls
-        load() later on in a way that makes sense to your code.
-
+        :meth:`load` later on in a way that makes sense to your code.
         """
         pass
 
     @classmethod
     @abstractmethod
     def load(cls) -> Any:
-        """
-        Load this thing from somewhere and give back an instance which
-        will become the global singleton and which will may (see
-        below) be save()d at program exit time.
+        """Load this thing from somewhere and give back an instance which
+        will become the global singleton and which may (see
+        below) be saved (via :meth:`save`) at program exit time.
 
-        Oh, in case this is handy, here's how to write a factory
-        method that doesn't call the c'tor in python:
+        Oh, in case this is handy, here's a reminder how to write a
+        factory method that doesn't call the c'tor in python::
 
             @classmethod
             def load_from_somewhere(cls, somewhere):
@@ -52,19 +55,40 @@ class Persistent(ABC):
                 # Load the piece(s) of obj that you want to from somewhere.
                 obj._state = load_from_somewhere(somewhere)
                 return obj
-
         """
         pass
 
 
 def was_file_written_today(filename: str) -> bool:
-    """Returns True if filename was written today."""
+    """Convenience wrapper around :meth:`was_file_written_within_n_seconds`.
+
+    Args:
+        filename: filename to check
+
+    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
+    assert mtime is not None
     now = datetime.datetime.now()
     return mtime.month == now.month and mtime.day == now.day and mtime.year == now.year
 
@@ -73,15 +97,37 @@ def was_file_written_within_n_seconds(
     filename: str,
     limit_seconds: int,
 ) -> bool:
-    """Returns True if filename was written within the pas limit_seconds
-    seconds.
-
+    """Helper for determining persisted state staleness.
+
+    Args:
+        filename: the filename to check
+        limit_seconds: how fresh, in seconds, it must be
+
+    Returns:
+        True if filename was written within the past limit_seconds
+        or False otherwise (or on error).
+
+    >>> 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
+    assert mtime is not None
     now = datetime.datetime.now()
     return (now - mtime).total_seconds() <= limit_seconds
 
@@ -89,8 +135,14 @@ def was_file_written_within_n_seconds(
 class PersistAtShutdown(enum.Enum):
     """
     An enum to describe the conditions under which state is persisted
-    to disk.  See details below.
-
+    to disk.  This is passed as an argument to the decorator below and
+    is used to indicate when to call :meth:`save` on a :class:`Persistent`
+    subclass.
+
+    * NEVER: never call :meth:`save`
+    * IF_NOT_LOADED: call :meth:`save` as long as we did not successfully
+      :meth:`load` its state.
+    * ALWAYS: always call :meth:`save`
     """
 
     NEVER = (0,)
@@ -99,23 +151,31 @@ class PersistAtShutdown(enum.Enum):
 
 
 class persistent_autoloaded_singleton(object):
-    """A decorator that can be applied to a Persistent subclass (i.e.  a
-    class with a save() and load() method.  It will intercept attempts
-    to instantiate the class via it's c'tor and, instead, invoke the
-    class' load() method to give it a chance to read state from
-    somewhere persistent.
-
-    If load() fails (returns None), the c'tor is invoked with the
+    """A decorator that can be applied to a :class:`Persistent` subclass
+    (i.e.  a class with :meth:`save` and :meth:`load` methods.  The
+    decorator will intercept attempts to instantiate the class via
+    it's c'tor and, instead, invoke the class' :meth:`load` to give it a
+    chance to read state from somewhere persistent (disk, db,
+    whatever).  Subsequent calls to construt instances of the wrapped
+    class will return a single, global instance (i.e. the wrapped
+    class is a singleton).
+
+    If :meth:`load` fails (returns None), the c'tor is invoked with the
     original args as a fallback.
 
-    Based upon the value of the optional argument persist_at_shutdown,
-    (NEVER, IF_NOT_LOADED, ALWAYS), the save() method of the class will
-    be invoked just before program shutdown to give the class a chance
-    to save its state somewhere.
+    Based upon the value of the optional argument
+    :code:`persist_at_shutdown` argument, (NEVER, IF_NOT_LOADED,
+    ALWAYS), the :meth:`save` method of the class will be invoked just
+    before program shutdown to give the class a chance to save its
+    state somewhere.
 
-    The implementations of save() and load() and where the class
-    persists its state are details left to the Persistent
-    implementation.
+    .. note::
+        The implementations of :meth:`save` and :meth:`load` and where the
+        class persists its state are details left to the :class:`Persistent`
+        implementation.  Essentially this decorator just handles the
+        plumbing of calling your save/load and appropriate times and
+        creates a transparent global singleton whose state can be
+        persisted between runs.
 
     """
 
@@ -136,35 +196,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()