Source code for edri.utility.cache

from collections import defaultdict
from datetime import datetime, timedelta
from pytz import timezone

__all__ = ["Cache"]




class Cache:
    timezone = timezone("Europe/Prague")

    def __init__(self) -> None:
        self.last_modified: defaultdict[str, datetime] = defaultdict(lambda: datetime.now(tz=self.timezone))

    def __len__(self) -> int:
        return len(self.last_modified)



    def last_change(self, key: str) -> datetime:
        return self.last_modified[key]




    def tag(self, key: str) -> str:
        # Microsecond resolution: a plain integer-seconds tag would collide for
        # two changes in the same second, letting a stale 304 slip through.
        # renew() guarantees last_change() is strictly increasing per microsecond.
        dt = self.last_change(key)
        return f"{key}-{int(dt.timestamp())}{dt.microsecond:06d}"




    def seed(self, key: str, when: datetime) -> None:
        """Set the last-change timestamp for ``key`` explicitly.

        Use from ``Manager.after_start()`` to pre-seed a cache key so its first
        ETag reflects a known baseline instead of process start time. A naive
        ``when`` is assumed to be in the cache timezone, keeping it comparable
        with the aware timestamps ``renew()`` produces.
        """
        if when.tzinfo is None:
            when = when.replace(tzinfo=self.timezone)
        self.last_modified[key] = when




    def renew(self, key: str) -> None:
        now = datetime.now(tz=self.timezone)
        previous = self.last_modified.get(key)
        # datetime.now() has finite resolution: two renews close enough in
        # time (e.g. under CPU contention) can land on the same instant,
        # which would break callers relying on last_change() strictly
        # increasing across renewals.
        if previous is not None and now <= previous:
            now = previous + timedelta(microseconds=1)
        self.last_modified[key] = now