151 lines
4.8 KiB
Python
151 lines
4.8 KiB
Python
from .internal import assert_subpackage; assert_subpackage(__name__)
|
|
|
|
from contextlib import contextmanager
|
|
import logging
|
|
from pathlib import Path
|
|
import sys
|
|
from typing import Optional, Iterator, cast, TYPE_CHECKING, TypeVar, Callable, overload, Union, Any, Type
|
|
import warnings
|
|
|
|
import appdirs # type: ignore[import-untyped]
|
|
|
|
PathIsh = Union[str, Path] # avoid circular import from .common
|
|
|
|
|
|
def disable_cachew() -> None:
|
|
try:
|
|
import cachew # noqa: F401 # unused, it's fine
|
|
except ImportError:
|
|
# nothing to disable
|
|
return
|
|
|
|
from cachew import settings
|
|
|
|
settings.ENABLE = False
|
|
|
|
|
|
@contextmanager
|
|
def disabled_cachew() -> Iterator[None]:
|
|
try:
|
|
import cachew # noqa: F401 # unused, it's fine
|
|
except ImportError:
|
|
# nothing to disable
|
|
yield
|
|
return
|
|
from cachew.extra import disabled_cachew
|
|
|
|
with disabled_cachew():
|
|
yield
|
|
|
|
|
|
def _appdirs_cache_dir() -> Path:
|
|
cd = Path(appdirs.user_cache_dir('my'))
|
|
cd.mkdir(exist_ok=True, parents=True)
|
|
return cd
|
|
|
|
|
|
_CACHE_DIR_NONE_HACK = Path('/tmp/hpi/cachew_none_hack')
|
|
|
|
|
|
def cache_dir(suffix: Optional[PathIsh] = None) -> Path:
|
|
from . import core_config as CC
|
|
|
|
cdir_ = CC.config.get_cache_dir()
|
|
|
|
sp: Optional[Path] = None
|
|
if suffix is not None:
|
|
sp = Path(suffix)
|
|
# guess if you do need absolute, better path it directly instead of as suffix?
|
|
assert not sp.is_absolute(), sp
|
|
|
|
# ok, so ideally we could just return cdir_ / sp
|
|
# however, this function was at first used without the suffix, e.g. cache_dir() / 'some_dir'
|
|
# but now cache_dir setting can also be None which means 'disable cache'
|
|
# changing return type to Optional means that it will break for existing users even if the cache isn't used
|
|
# it's kinda wrong.. so we use dummy path (_CACHE_DIR_NONE_HACK), and then strip it away in core.common.mcachew
|
|
# this logic is tested via test_cachew_dir_none
|
|
|
|
if cdir_ is None:
|
|
cdir = _CACHE_DIR_NONE_HACK
|
|
else:
|
|
cdir = cdir_
|
|
|
|
return cdir if sp is None else cdir / sp
|
|
|
|
|
|
"""See core.cachew.cache_dir for the explanation"""
|
|
|
|
|
|
_cache_path_dflt = cast(str, object())
|
|
|
|
|
|
# TODO I don't really like 'mcachew', just 'cache' would be better... maybe?
|
|
# todo ugh. I think it needs @doublewrap, otherwise @mcachew without args doesn't work
|
|
# but it's a bit problematic.. doublewrap works by defecting if the first arg is callable
|
|
# but here cache_path can also be a callable (for lazy/dynamic path)... so unclear how to detect this
|
|
def _mcachew_impl(cache_path=_cache_path_dflt, **kwargs):
|
|
"""
|
|
Stands for 'Maybe cachew'.
|
|
Defensive wrapper around @cachew to make it an optional dependency.
|
|
"""
|
|
if cache_path is _cache_path_dflt:
|
|
# wasn't specified... so we need to use cache_dir
|
|
cache_path = cache_dir()
|
|
|
|
if isinstance(cache_path, (str, Path)):
|
|
try:
|
|
# check that it starts with 'hack' path
|
|
Path(cache_path).relative_to(_CACHE_DIR_NONE_HACK)
|
|
except: # noqa: E722 bare except
|
|
pass # no action needed, doesn't start with 'hack' string
|
|
else:
|
|
# todo show warning? tbh unclear how to detect when user stopped using 'old' way and using suffix instead?
|
|
# if it does, means that user wanted to disable cache
|
|
cache_path = None
|
|
try:
|
|
import cachew
|
|
except ModuleNotFoundError:
|
|
warnings.warn('cachew library not found. You might want to install it to speed things up. See https://github.com/karlicoss/cachew')
|
|
return lambda orig_func: orig_func
|
|
else:
|
|
kwargs['cache_path'] = cache_path
|
|
return cachew.cachew(**kwargs)
|
|
|
|
|
|
if TYPE_CHECKING:
|
|
R = TypeVar('R')
|
|
if sys.version_info[:2] >= (3, 10):
|
|
from typing import ParamSpec
|
|
else:
|
|
from typing_extensions import ParamSpec
|
|
P = ParamSpec('P')
|
|
CC = Callable[P, R] # need to give it a name, if inlined into bound=, mypy runs in a bug
|
|
PathProvider = Union[PathIsh, Callable[P, PathIsh]]
|
|
# NOTE: in cachew, HashFunction type returns str
|
|
# however in practice, cachew alwasy calls str for its result
|
|
# so perhaps better to switch it to Any in cachew as well
|
|
HashFunction = Callable[P, Any]
|
|
|
|
F = TypeVar('F', bound=Callable)
|
|
|
|
# we need two versions due to @doublewrap
|
|
# this is when we just annotate as @cachew without any args
|
|
@overload # type: ignore[no-overload-impl]
|
|
def mcachew(fun: F) -> F:
|
|
...
|
|
|
|
@overload
|
|
def mcachew(
|
|
cache_path: Optional[PathProvider] = ...,
|
|
*,
|
|
force_file: bool = ...,
|
|
cls: Optional[Type] = ...,
|
|
depends_on: HashFunction = ...,
|
|
logger: Optional[logging.Logger] = ...,
|
|
chunk_by: int = ...,
|
|
synthetic_key: Optional[str] = ...,
|
|
) -> Callable[[F], F]:
|
|
...
|
|
|
|
else:
|
|
mcachew = _mcachew_impl
|