Add a simple test for waitable presence.

[python_utils.git] / persistent.py

diff --git a/persistent.py b/persistent.py
index 5c2b132448ebce64271c0165a8848337db5e69c8..119931b8ccba607ccc48321ac6f3d6dd3dd5b791 100644 (file)
--- 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,12 +64,9 @@ 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(
@@ -84,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
 
@@ -130,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
@@ -144,32 +142,23 @@ class persistent_autoloaded_singleton(object):
 
             # Otherwise, try to load it from persisted state.
             was_loaded = False
-            logger.debug(
-                f'Attempting to load {cls.__name__} from persisted state.'
-            )
+            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'Attempting to instantiate {cls.__name__} directly.'
-                )
+                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