#!/usr/bin/env python3
-from abc import ABC, abstractmethod
+# © 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
import functools
import logging
-from typing import Callable, Optional
-
-import dill
+from abc import ABC, abstractmethod
+from typing import Any
import file_utils
class Persistent(ABC):
"""
- A base class of an object with a load/save method.
+ A base class of an object with a load/save method. Classes that are
+ decorated with :code:`@persistent_autoloaded_singleton` should subclass
+ this and implement their :meth:`save` and :meth:`load` methods.
"""
+
@abstractmethod
- def save(self):
+ def save(self) -> bool:
+ """
+ Save this thing somewhere that you'll remember when someone calls
+ :meth:`load` later on in a way that makes sense to your code.
+ """
pass
+ @classmethod
@abstractmethod
- def load(self):
+ def load(cls) -> Any:
+ """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 a reminder how to write a
+ factory method that doesn't call the c'tor in python::
+
+ @classmethod
+ def load_from_somewhere(cls, somewhere):
+ # Note: __new__ does not call __init__.
+ obj = cls.__new__(cls)
+
+ # Don't forget to call any polymorphic base class initializers
+ super(MyClass, obj).__init__()
+
+ # Load the piece(s) of obj that you want to from somewhere.
+ obj._state = load_from_somewhere(somewhere)
+ return obj
+ """
pass
-def reuse_if_mtime_is_today() -> Callable[[datetime.datetime], bool]:
+def was_file_written_today(filename: str) -> bool:
+ """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
"""
- A helper that returns a lambda appropriate for use in the
- persistent_autoloaded_singleton decorator's may_reuse_persisted
- parameter that allows persisted state to be reused as long as it
- was persisted on the same day as the load.
- """
- now = datetime.datetime.now()
- return lambda dt: (
- dt.month == now.month and
- dt.day == now.day and
- dt.year == now.year
- )
-
-
-def reuse_if_mtime_less_than_limit_sec(
- limit_seconds: int
-) -> Callable[[datetime.datetime], bool]:
- """
- A helper that returns a lambda appropriate for use in the
- persistent_autoloaded_singleton decorator's may_reuse_persisted
- parameter that allows persisted state to be reused as long as it
- was persisted within the past limit_seconds.
+ 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 lambda dt: (now - dt).total_seconds() <= limit_seconds
+ return mtime.month == now.month and mtime.day == now.day and mtime.year == now.year
+
+
+def was_file_written_within_n_seconds(
+ filename: str,
+ limit_seconds: int,
+) -> bool:
+ """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
-def dont_reuse_persisted_state_force_refresh(
-) -> Callable[[datetime.datetime], bool]:
- return lambda dt: 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
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,
- IF_NOT_INITIALIZED_FROM_DISK = 1,
- ALWAYS = 2,
-
-
-class persistent_autoloaded_singleton(Persistent):
- """This class is meant to be used as a decorator around a class that:
-
- 1. Is a singleton; one global instance per python program.
- 2. Has a complex state that is initialized fully by __init__()
- 3. Would benefit from caching said state on disk and reloading
- it on future invokations rather than recomputing and
- reinitializing.
-
- Here's and example usage pattern:
-
- @persistent_autoloaded_singleton(
- filename = "my_cache_file.bin",
- may_reuse_persisted = reuse_if_mtime_less_than_limit_sec(60),
- persist_at_shutdown = PersistAtShutdown.IF_NOT_INITIALIZED_FROM_DISK,
- )
- class MyComplexObject(object):
- def __init__(self, ...):
- # do a bunch of work to fully initialize this instance
-
- def another_method(self, ...):
- # use the state stored in this instance to do some work
-
- What does this do, exactly?
-
- 1. Anytime you attempt to instantiate MyComplexObject you will
- get the same instance. This class is now a singleton.
- 2. The first time you attempt to instantiate MyComplexObject
- the wrapper scaffolding will check "my_cache_file.bin". If
- it exists and any may_reuse_persisted predicate indicates
- that reusing persisted state is allowed, we will skip the
- call to __init__ and return an unpickled instance read from
- the disk file. In the example above the predicate allows
- reuse of saved state if it is <= 60s old.
- 3. If the file doesn't exist or the predicate indicates that
- the persisted state cannot be reused (e.g. too stale),
- MyComplexObject's __init__ will be invoked and will be
- expected to fully initialize the instance.
- 4. At program exit time, depending on the value of the
- persist_at_shutdown parameter, the state of MyComplexObject
- will be written to disk using the same filename so that
- future instances may potentially reuse saved state. Note
- that the state that is persisted is the state at program
- exit time. In the example above this parameter indicates
- that we should persist state so long as we were not
- initialized from cached state on disk.
+
+ NEVER = (0,)
+ IF_NOT_LOADED = (1,)
+ ALWAYS = (2,)
+
+
+class persistent_autoloaded_singleton(object):
+ """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
+ :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.
+
+ .. 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.
"""
+
def __init__(
- self,
- filename: str,
- *,
- may_reuse_persisted: Optional[Callable[[datetime.datetime], bool]] = None,
- persist_at_shutdown: PersistAtShutdown = PersistAtShutdown.NEVER):
- self.filename = filename
- self.may_reuse_persisted = may_reuse_persisted
+ self,
+ *,
+ persist_at_shutdown: PersistAtShutdown = PersistAtShutdown.IF_NOT_LOADED,
+ ):
self.persist_at_shutdown = persist_at_shutdown
self.instance = None
- def __call__(self, cls):
- @functools.wraps(cls)
+ def __call__(self, cls: Persistent):
+ @functools.wraps(cls) # type: ignore
def _load(*args, **kwargs):
# If class has already been loaded, act like a singleton
# 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
- was_loaded_from_disk = False
- if file_utils.does_file_exist(self.filename):
- cache_mtime_dt = file_utils.get_file_mtime_as_datetime(
- self.filename
- )
- now = datetime.datetime.now()
- if (
- self.may_reuse_persisted is not None and
- self.may_reuse_persisted(cache_mtime_dt)
- ):
- logger.debug(
- f'Attempting to load from persisted cache (mtime={cache_mtime_dt}, {now-cache_mtime_dt} ago)')
- if not self.load():
- logger.warning('Loading from cache failed?!')
- assert self.instance is None
- else:
- assert self.instance is not None
- was_loaded_from_disk = True
-
- if self.instance is None:
- logger.debug(
- f'Attempting to instantiate {cls.__name__} directly.'
- )
+ # Otherwise, try to load it from persisted state.
+ was_loaded = False
+ 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('Attempting to instantiate %s directly.', cls.__name__)
self.instance = cls(*args, **kwargs)
- was_loaded_from_disk = False
+ else:
+ 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_from_disk and
- self.persist_at_shutdown is PersistAtShutdown.IF_NOT_INITIALIZED_FROM_DISK
- )
+
+ if self.persist_at_shutdown is PersistAtShutdown.ALWAYS or (
+ not was_loaded and self.persist_at_shutdown is PersistAtShutdown.IF_NOT_LOADED
):
- atexit.register(self.save)
+ logger.debug('Scheduling a deferred called to save at process shutdown time.')
+ atexit.register(self.instance.save)
return self.instance
+
return _load
- def load(self) -> bool:
- 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
+if __name__ == '__main__':
+ import doctest
+
+ doctest.testmod()
projects / python_utils.git / blobdiff