Source code for temporalcache.utils

# *****************************************************************************
#
# Copyright (c) 2021, the temporal-cache authors.
#
# This file is part of the temporal-cache library, distributed under the terms of
# the Apache License 2.0.  The full license can be found in the LICENSE file.
#
import os
from abc import ABCMeta, abstractmethod
from functools import lru_cache, wraps

TEMPORAL_CACHE_GLOBAL_DISABLE = os.environ.get("TEMPORAL_CACHE_DISABLE", False)


[docs]def disable(): global TEMPORAL_CACHE_GLOBAL_DISABLE TEMPORAL_CACHE_GLOBAL_DISABLE = True
[docs]def enable(): global TEMPORAL_CACHE_GLOBAL_DISABLE TEMPORAL_CACHE_GLOBAL_DISABLE = False
[docs]class TCException(Exception): pass
[docs]class StorageBase(metaclass=ABCMeta):
[docs] @abstractmethod def cache_clear(self): pass
def __call__(self, foo): """No caching by default""" @wraps(foo) def _wrapper(*args, **kwargs): return foo(*args, **kwargs) return _wrapper
def _base(last, now, lap, offset, multiple, attr): """ last - last datetime now - current datetime lap - at what point to "roll over" offset - how many seconds between laps attr - what to look at multiple - what to multiply attr by to get seconds """ # last started before :X, so if now > :X # handle windows timestamp issues try: now_ts = now.timestamp() except OSError: now_ts = 0 try: last_ts = last.timestamp() except OSError: last_ts = 0 diff = round(now_ts, 0) - round(last_ts, 0) min_gap = offset - getattr(last, attr) * multiple if diff > offset: return True if getattr(last, attr) < lap: if getattr(now, attr) >= lap: return True return False # last started after :X, so if now > last + interval, or if now > :X elif getattr(now, attr) >= lap and diff >= min_gap: return True return False def _secondly(last, now, secondly): return _base(last=last, now=now, lap=secondly, offset=60, multiple=1, attr="second") def _minutely(last, now, minutely): return _base( last=last, now=now, lap=minutely, offset=3600, multiple=60, attr="minute" ) def _hourly(last, now, hourly, tz=None): return _base( last=last, now=now, lap=hourly, offset=3600 * 24, multiple=3600, attr="hour" ) def _daily(last, now, daily, tz=None): return _base( last=last, now=now, lap=daily, offset=3600 * 24 * 30, multiple=3600 * 24, attr="day", ) # FIXME day should be derived from calendar def _day_of_week(last, now, day_of_week, tz=None): return _base( last=last, now=now, lap=day_of_week, offset=3600 * 24 * 7, multiple=3600 * 24, attr="day", ) def _weekly(last, now, weekly, tz=None): return _base( last=last, now=now, lap=weekly, offset=3600 * 24 * 7 * 4.34, multiple=3600 * 24 * 7, attr="week", ) # FIXME # of weeks should be derived from calendar def _monthly(last, now, monthly, tz=None): return _base( last=last, now=now, lap=monthly, offset=3600 * 24 * 365, multiple=3600 * 24 * 7 * 4.34, attr="month", ) # FIXME # of weeks should be derived from calendar
[docs]def should_expire( last, now, secondly=None, minutely=None, hourly=None, daily=None, day_of_week=None, weekly=None, monthly=None, ): """should the cache expire? last - datetime now - datetime if yearly: necessary_distance = calc(0, 0, 0, 0, 0, 0, yearly) """ sec_res = _secondly(last, now, secondly) if secondly is not None else True min_res = _minutely(last, now, minutely) if minutely is not None else True hou_res = _hourly(last, now, hourly) if hourly is not None else True dow_res = _day_of_week(last, now, day_of_week) if day_of_week is not None else True dai_res = _daily(last, now, daily) if daily is not None else True wee_res = _weekly(last, now, weekly) if weekly is not None else True mon_res = _monthly(last, now, monthly) if weekly is not None else True return all((sec_res, min_res, hou_res, dai_res, dow_res, wee_res, mon_res))
[docs]@lru_cache(1000) def calc(seconds=0, minutes=0, hours=0, days=0, weeks=0, months=0, years=0): return ( seconds + minutes * 60 + hours * 60 * 60 + days * 24 * 60 * 60 + weeks * 7 * 24 * 60 * 60 + months * 30 * 7 * 24 * 60 * 60 + years * 365 * 24 * 60 * 60 )