Saving and reading the last successful execution as a json-file

Question

I've got some tools that run perdiodically like weekly or monthly that don't support any retry mechanism. In order to increase their chance to succeed in case a database wasn't reachable or some other problem occured, I simply schedule them to run several times per day or a couple of days in month.

Currently they do their job every time they run, regardles whether they succeeded the first time or not. I'd like to change this by saving a simple token with the timestamp of the last successful execution.

---

Implementation

This is where my Tick class comes into play. I designed it handle a json-token with only a single property. I know at this point it could be a plain text file, but this escalates pretty quickly so I'd like it to be future-proof.

{"created_on": "2022年09月01日T15:36:08.434008"}

It allows me to save the current datetime and also takes care of reading it and checking whether the token is still valid or already expired. Its job is simple, but I wanted to have it as pretty and as smartass as it can be. This is how it looks like:

import json
import pathlib
import os
from functools import lru_cache
from datetime import datetime, date
from typing import Any, Callable, Dict, Union
class Tick:
 def __init__(self, file_name: Union[str, bytes, os.PathLike], lifetime: float, calc_as: Callable[..., float]):
 self.file_name = file_name
 self.lifetime = lifetime
 self.calc_as = calc_as
 def check(self):
 pathlib.Path(os.path.dirname(self.file_name)).mkdir(parents=True, exist_ok=True)
 data = json.dumps({self.created_on.__name__: datetime.now()}, cls=_JsonDateTimeEncoder)
 with open(self.file_name, "w") as f:
 f.write(data)
 self.created_on.cache_clear()
 @lru_cache # avoid reading the file more than once until updated
 def created_on(self) -> Union[datetime, None]:
 if not os.path.isfile(self.file_name):
 return None
 with open(self.file_name, "r") as f:
 return json.loads(f.read(), cls=_JsonDateTimeDecoder)[self.created_on.__name__]
 def seconds(self) -> Union[float, None]:
 return (datetime.now() - self.created_on()).total_seconds() if self.created_on() else None
 def minutes(self) -> Union[float, None]:
 return self.seconds() / 60 if self.seconds() else None
 def hours(self) -> Union[float, None]:
 return self.minutes() / 60 if self.minutes() else None
 def days(self) -> Union[float, None]:
 return self.hours() / 24 if self.hours() else None
 def is_expired(self) -> bool:
 return not self.calc_as(self) or self.calc_as(self) > self.lifetime
 def is_valid(self) -> bool:
 return not self.is_expired()

Since the json package cannot handle datetime by default, I created these two classes to handle that part.

class _JsonDateTimeEncoder(json.JSONEncoder):
 def default(self, o: Any) -> Any:
 if isinstance(o, (date, datetime)):
 return o.isoformat()
class _JsonDateTimeDecoder(json.JSONDecoder):
 def __init__(self):
 super().__init__(object_hook=self.parse_datetime_or_default)
 @staticmethod
 def parse_datetime_or_default(d: Dict):
 r = dict()
 for k in d.keys():
 r[k] = d[k]
 if isinstance(d[k], str):
 try:
 r[k] = datetime.fromisoformat(d[k]) # try parse date-time
 except ValueError:
 pass # default value is already set
 return r

---

Example

This is how I test it:

import json
from datetime import datetime
from src.tick_off import Tick
if __name__ == "__main__":
 tick = Tick(r"c:\temp\token_test__.json", lifetime=5, calc_as=Tick.hours)
 # tick.check()
 print(tick.hours())
 print("is_expired:", tick.is_expired())
 print("is_valid:", tick.is_valid())

---

The one thing that bothers me the most is the repeated conditions in methods like minutes, hours & days that use virtually identical conditions:

return self.seconds() / 60 if self.seconds() else None

Unfortunatelly None / 60 doesn't produce None but an error, but maybe there is some smart way one can get rid of these conditions without turning it into a try/except that would be even more ugly?

Answer 1

First the positives: you've put some effort into typehints, caching, and encapsulation, all good things. You've started into pathlib support, which is good, and you target Python 3, which is good.

Union[datetime, None] is just Optional[datetime].

file_name is too broad in its type. Just force the caller to pass a Path. This is not an external-facing library: you don't need to, and should not, adopt the shotgun approach to argument types. After all of that effort, you still don't use Path properly - you're calling top-level open() and os.path.isfile when you should be using the Path equivalents.

When you write

pathlib.Path(self.file_name).mkdir(parents=True, exist_ok=True)

It seems to me like this should be getting .parent, not making a directory named the same as the target file.

calc_as is a fascinating, convoluted mistake. Don't do any of this. Just accept a timedelta for lifetime instead of a float. Delete all of your time unit conversion functions.

Don't dumps and loads string-based functions; call their file alternatives dump and load.

is_expired etc. are good candidates for being made @propertys.

If you want to support datetime parsing in JSON, fine. Your implementation, first, is more complicated than it needs to be: don't subclass JSONEncoder/JSONDecoder; just pass function references to object_hook and default at time of serialisation. Also, parse_datetime_or_default has been written to be too generic. Think: it attempts to convert any value in JSON to a datetime, regardless of whether that's appropriate. You have a fixed dictionary format; adhere to it.

self.created_on.__name__ is not necessary. You know the key name, so just... write it as a string.

Write unit tests.

check is a deceptive name. It doesn't check anything; it creates the file.

Don't lru_cache; just cache.

Suggested

import json
from functools import cache
from datetime import datetime, date, timedelta
from pathlib import Path
from tempfile import NamedTemporaryFile
from time import sleep
from typing import Any
def json_default(o: Any) -> Any:
 if isinstance(o, (date, datetime)):
 return o.isoformat()
def json_object_hook(d: dict[str, Any]) -> dict[str, Any]:
 return {
 **d,
 'created_on': datetime.fromisoformat(d['created_on']),
 }
class Tick:
 def __init__(self, file_name: Path, lifetime: timedelta) -> None:
 self.file_name = file_name
 self.lifetime = lifetime
 def create(self) -> None:
 self.file_name.parent.mkdir(parents=True, exist_ok=True)
 data = {'created_on': datetime.now()}
 with self.file_name.open('w') as f:
 json.dump(data, f, default=json_default)
 self.get_created_on.cache_clear()
 @property
 def is_created(self) -> bool:
 return self.file_name.is_file()
 @cache # avoid reading the file more than once until updated
 def get_created_on(self) -> datetime:
 with self.file_name.open() as f:
 return json.load(f, object_hook=json_object_hook)['created_on']
 @property
 def elapsed(self) -> timedelta:
 return datetime.now() - self.get_created_on()
 @property
 def is_valid(self) -> bool:
 return self.is_created and self.elapsed < self.lifetime
 @property
 def is_expired(self) -> bool:
 return not self.is_valid
def test() -> None:
 lifetime = timedelta(seconds=1)
 tick = Tick(Path('noexist'), lifetime)
 assert not tick.is_created
 assert not tick.is_valid
 assert tick.is_expired
 with NamedTemporaryFile(suffix='_token_test.json') as temp:
 tick = Tick(Path(temp.name), lifetime)
 assert tick.is_created
 tick.create()
 assert tick.is_created
 assert tick.is_valid
 assert not tick.is_expired
 sleep(1.5)
 assert tick.is_created
 assert not tick.is_valid
 assert tick.is_expired
if __name__ == '__main__':
 test()

• t3chb0t: Damn! This is clean as a whistle now; especially with the timedelta! I also love the new test code that I need practice more writing like that.

Answer 2

This self-answer might not entirely be on-topic, but I think it's worth mentioning that NamedTemporaryFile won't work on Windows as it keeps the file open while the Tick class will try to open it again and fail. Only Linux can handle this.

In order to make the suggusted test work on Windows too, I wrote a similar class that releases the file after creation and provides its name:

class TemporaryFileName:
 def __init__(self, suffix: str, create=False, delete=True):
 self.suffix = suffix
 self.create = create
 self.delete = delete
 def __enter__(self) -> Path:
 self._temp_name = Path(os.path.join(tempfile.gettempdir(), f"{os.urandom(4).hex()}_{self.suffix}"))
 try:
 return self._temp_name
 finally:
 if self.create:
 self._temp_name.open("x").close()
 def __exit__(self, exc_type, exc_val, exc_tb):
 if self.delete and self._temp_name.is_file():
 os.remove(self._temp_name)
```