#!/usr/bin/env python3 # © 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 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. """ @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. """ 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. Oh, in case this is handy, here's 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: """Returns True if filename was written today.""" 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: """Returns True if filename was written within the pas limit_seconds 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 (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. """ NEVER = (0,) IF_NOT_LOADED = (1,) ALWAYS = (2,) 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 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. The implementations of save() and load() and where the class persists its state are details left to the Persistent implementation. """ 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