projects
/
python_utils.git
/ blobdiff
commit
grep
author
committer
pickaxe
?
search:
re
summary
|
shortlog
|
log
|
commit
|
commitdiff
|
tree
raw
|
inline
| side by side
How to do the thing.
[python_utils.git]
/
persistent.py
diff --git
a/persistent.py
b/persistent.py
index f6ca0a0faa4bde09299db88d100fd1e9b8640f0a..119931b8ccba607ccc48321ac6f3d6dd3dd5b791 100644
(file)
--- a/
persistent.py
+++ b/
persistent.py
@@
-1,11
+1,11
@@
#!/usr/bin/env python3
#!/usr/bin/env python3
-from abc import ABC, abstractmethod
import atexit
import datetime
import enum
import functools
import logging
import atexit
import datetime
import enum
import functools
import logging
+from abc import ABC, abstractmethod
from typing import Any
import file_utils
from typing import Any
import file_utils
@@
-20,6
+20,7
@@
class Persistent(ABC):
and implement their save() and load() methods.
"""
and implement their save() and load() methods.
"""
+
@abstractmethod
def save(self) -> bool:
"""
@abstractmethod
def save(self) -> bool:
"""
@@
-63,17
+64,14
@@
def was_file_written_today(filename: str) -> bool:
return False
mtime = file_utils.get_file_mtime_as_datetime(filename)
return False
mtime = file_utils.get_file_mtime_as_datetime(filename)
+ assert mtime is not None
now = datetime.datetime.now()
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(
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.
) -> bool:
"""Returns True if filename was written within the pas limit_seconds
seconds.
@@
-83,6
+81,7
@@
def was_file_written_within_n_seconds(
return False
mtime = file_utils.get_file_mtime_as_datetime(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
now = datetime.datetime.now()
return (now - mtime).total_seconds() <= limit_seconds
@@
-93,9
+92,10
@@
class PersistAtShutdown(enum.Enum):
to disk. See details below.
"""
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):
class persistent_autoloaded_singleton(object):
@@
-118,15
+118,17
@@
class persistent_autoloaded_singleton(object):
implementation.
"""
implementation.
"""
+
def __init__(
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):
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
def _load(*args, **kwargs):
# If class has already been loaded, act like a singleton
@@
-143,7
+145,8
@@
class persistent_autoloaded_singleton(object):
logger.debug(f'Attempting to load {cls.__name__} from persisted state.')
self.instance = cls.load()
if not self.instance:
logger.debug(f'Attempting to load {cls.__name__} from persisted state.')
self.instance = cls.load()
if not self.instance:
- logger.warning('Loading from cache failed.')
+ msg = 'Loading from cache failed.'
+ logger.warning(msg)
logger.debug(f'Attempting to instantiate {cls.__name__} directly.')
self.instance = cls(*args, **kwargs)
else:
logger.debug(f'Attempting to instantiate {cls.__name__} directly.')
self.instance = cls(*args, **kwargs)
else:
@@
-152,14
+155,11
@@
class persistent_autoloaded_singleton(object):
assert self.instance is not None
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
):
logger.debug('Scheduling a deferred called to save at process shutdown time.')
atexit.register(self.instance.save)
return self.instance
+
return _load
return _load