3
\$\begingroup\$

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-01T15: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?

\$\endgroup\$

2 Answers 2

3
\$\begingroup\$

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()
\$\endgroup\$
1
  • 2
    \$\begingroup\$ 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. \$\endgroup\$
    – t3chb0t
    Sep 2 at 15:15
0
\$\begingroup\$

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)
```
\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.