X-Git-Url: https://wannabe.guru.org/gitweb/?a=blobdiff_plain;f=persistent.py;h=5c2b132448ebce64271c0165a8848337db5e69c8;hb=36fea7f15ed17150691b5b3ead75450e575229ef;hp=30e4ccbfcbed724ead5445b9674a2334d621497c;hpb=c46018e2b1ddc78f9df557c3fb24d2c2c849f054;p=python_utils.git diff --git a/persistent.py b/persistent.py index 30e4ccb..5c2b132 100644 --- a/persistent.py +++ b/persistent.py @@ -2,34 +2,134 @@ from abc import ABC, abstractmethod import atexit +import datetime +import enum import functools import logging - -import dill +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): + 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(self): + 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 -class persistent_autoload_singleton(Persistent): - def __init__(self, filename: str, *, max_age_sec: int = 0): - self.filename = filename - self.max_age_sec = max_age_sec +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) + 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) + 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): + def __call__(self, cls: Persistent): @functools.wraps(cls) def _load(*args, **kwargs): @@ -42,48 +142,35 @@ class persistent_autoload_singleton(Persistent): ) return self.instance - if not self.load(): - assert self.instance is None + # Otherwise, try to load it from persisted state. + was_loaded = False + logger.debug( + f'Attempting to load {cls.__name__} from persisted state.' + ) + self.instance = cls.load() + if not self.instance: + msg = 'Loading from cache failed.' + logger.warning(msg) logger.debug( - f'Instantiating {cls.__name__} directly.' + f'Attempting to instantiate {cls.__name__} directly.' ) self.instance = cls(*args, **kwargs) + else: + logger.debug( + f'Class {cls.__name__} was loaded from persisted state successfully.' + ) + was_loaded = True - # On program exit, save state to disk. - atexit.register(self.save) assert self.instance is not None - return self.instance - return _load - def load(self) -> bool: - if ( - file_utils.does_file_exist(self.filename) - and ( - self.max_age_sec == 0 or - file_utils.get_file_mtime_age_seconds(self.filename) <= self.max_age_sec + 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.' ) - ): - logger.debug( - f'Attempting to load from file {self.filename}' - ) - try: - with open(self.filename, 'rb') as f: - self.instance = dill.load(f) - return True - except Exception: - self.instance = None - return False - return False + atexit.register(self.instance.save) + return self.instance - 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 + return _load