X-Git-Url: https://wannabe.guru.org/gitweb/?a=blobdiff_plain;f=persistent.py;h=119931b8ccba607ccc48321ac6f3d6dd3dd5b791;hb=65e6f781dffb268be6ef0a015f4cc89b57b62a5e;hp=d62dd6754eeffc78c1c09adea7c82e778f8450b3;hpb=e6f32fdd9b373dfcd100c7accb41f57d83c2f0a1;p=python_utils.git diff --git a/persistent.py b/persistent.py index d62dd67..119931b 100644 --- a/persistent.py +++ b/persistent.py @@ -1,11 +1,11 @@ #!/usr/bin/env python3 -from abc import ABC, abstractmethod import atexit import datetime import enum import functools import logging +from abc import ABC, abstractmethod from typing import Any import file_utils @@ -64,6 +64,7 @@ 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 @@ -80,6 +81,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 @@ -126,7 +128,7 @@ class persistent_autoloaded_singleton(object): 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 @@ -148,20 +150,15 @@ class persistent_autoloaded_singleton(object): logger.debug(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.' - ) + logger.debug(f'Class {cls.__name__} was loaded from persisted state successfully.') 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 + 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('Scheduling a deferred called to save at process shutdown time.') atexit.register(self.instance.save) return self.instance