#!/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 import functools import logging from abc import ABC, abstractmethod from typing import Any import file_utils logger = logging.getLogger(__name__) class Persistent(ABC): """ 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) -> 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(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 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 """ 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 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 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. 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_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, *, persist_at_shutdown: PersistAtShutdown = PersistAtShutdown.IF_NOT_LOADED, ): self.persist_at_shutdown = persist_at_shutdown self.instance = None def __call__(self, cls: Persistent): @functools.wraps(cls) # type: ignore 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( '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('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) 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 and self.persist_at_shutdown is PersistAtShutdown.IF_NOT_LOADED ): 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()