X-Git-Url: https://wannabe.guru.org/gitweb/?a=blobdiff_plain;f=persistent.py;h=c902313eb4a28cb4635779f584f8775a2b646d35;hb=e8fbbb7306430478dec55d2c963eed116d8330cc;hp=36ae29c9271eaaa678cb7f4d525b1c5d05ead0f8;hpb=52290f9c9e0eeaba3d5a067043f5ba98c9b386e5;p=python_utils.git diff --git a/persistent.py b/persistent.py index 36ae29c..c902313 100644 --- a/persistent.py +++ b/persistent.py @@ -1,13 +1,16 @@ #!/usr/bin/env python3 -from abc import ABC, abstractmethod +"""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 warnings import file_utils @@ -21,6 +24,7 @@ class Persistent(ABC): and implement their save() and load() methods. """ + @abstractmethod def save(self) -> bool: """ @@ -64,17 +68,14 @@ def was_file_written_today(filename: str) -> bool: 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 - ) + 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, + filename: str, + limit_seconds: int, ) -> bool: """Returns True if filename was written within the pas limit_seconds seconds. @@ -84,6 +85,7 @@ def was_file_written_within_n_seconds( 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 @@ -94,9 +96,10 @@ class PersistAtShutdown(enum.Enum): to disk. See details below. """ - NEVER = 0, - IF_NOT_LOADED = 1, - ALWAYS = 2, + + NEVER = (0,) + IF_NOT_LOADED = (1,) + ALWAYS = (2,) class persistent_autoloaded_singleton(object): @@ -119,15 +122,17 @@ class persistent_autoloaded_singleton(object): implementation. """ + def __init__( - self, - *, - persist_at_shutdown: PersistAtShutdown = PersistAtShutdown.IF_NOT_LOADED): + 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) + @functools.wraps(cls) # type: ignore def _load(*args, **kwargs): # If class has already been loaded, act like a singleton @@ -135,33 +140,30 @@ 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 - ) + 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