1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
|
#!/usr/bin/env python3
import atexit
import datetime
import enum
import functools
import logging
from abc import ABC, abstractmethod
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) -> 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(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
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)
assert mtime is not None
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)
assert mtime is not None
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: Persistent):
@functools.wraps(cls) # type: ignore
def _load(*args, **kwargs):
# If class has already been loaded, act like a singleton
# and return a reference to the one and only instance in
# memory.
if self.instance is not None:
logger.debug(
f'Returning already instantiated singleton instance of {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.')
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.')
self.instance = cls(*args, **kwargs)
else:
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
):
logger.debug('Scheduling a deferred called to save at process shutdown time.')
atexit.register(self.instance.save)
return self.instance
return _load
|
