forked from andreoliwa/nitpick
-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
feat: switch to requests-cache for style caching (andreoliwa#467)
Use `requests_cache.CachedSession()` for URL fetching and caching. file and pypackage styles are no longer cached but reuse the original file each time. This resolves issues with concurrent access to the cache, as it is now a sqlite database file.
- Loading branch information
Showing
8 changed files
with
145 additions
and
117 deletions.
There are no files selected for viewing
Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.
Oops, something went wrong.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -1,36 +1,44 @@ | ||
"""Cache functions and configuration for styles.""" | ||
from __future__ import annotations | ||
|
||
import re | ||
from datetime import timedelta | ||
from typing import Tuple | ||
|
||
from loguru import logger | ||
from requests_cache.cache_control import DO_NOT_CACHE, NEVER_EXPIRE | ||
|
||
from nitpick.enums import CachingEnum | ||
|
||
REGEX_CACHE_UNIT = re.compile(r"(?P<number>\d+)\s+(?P<unit>(minute|hour|day|week))", re.IGNORECASE) | ||
EXPIRES_DEFAULTS = { | ||
CachingEnum.NEVER: DO_NOT_CACHE, | ||
CachingEnum.FOREVER: NEVER_EXPIRE, | ||
CachingEnum.EXPIRES: timedelta(hours=1), | ||
} | ||
|
||
|
||
def parse_cache_option(cache_option: str) -> Tuple[CachingEnum, timedelta]: | ||
def parse_cache_option(cache_option: str) -> tuple[CachingEnum, timedelta | int]: | ||
"""Parse the cache option provided on pyproject.toml. | ||
If no cache if provided or is invalid, the default is *one hour*. | ||
""" | ||
clean_cache_option = cache_option.strip().lower() if cache_option else "" | ||
mapping = {CachingEnum.NEVER.name.lower(): CachingEnum.NEVER, CachingEnum.FOREVER.name.lower(): CachingEnum.FOREVER} | ||
simple_cache = mapping.get(clean_cache_option) | ||
if simple_cache: | ||
logger.info(f"Simple cache option: {simple_cache.name}") | ||
return simple_cache, timedelta() | ||
|
||
default = CachingEnum.EXPIRES, timedelta(hours=1) | ||
if not clean_cache_option: | ||
return default | ||
|
||
for match in REGEX_CACHE_UNIT.finditer(clean_cache_option): | ||
plural_unit = match.group("unit") + "s" | ||
number = int(match.group("number")) | ||
logger.info(f"Cache option with unit: {number} {plural_unit}") | ||
return CachingEnum.EXPIRES, timedelta(**{plural_unit: number}) | ||
|
||
logger.warning(f"Invalid cache option: {clean_cache_option}. Defaulting to 1 hour") | ||
return default | ||
clean_cache_option = cache_option.strip().upper() if cache_option else "" | ||
try: | ||
caching = CachingEnum[clean_cache_option] | ||
logger.info(f"Simple cache option: {caching.name}") | ||
except KeyError: | ||
caching = CachingEnum.EXPIRES | ||
|
||
expires_after = EXPIRES_DEFAULTS[caching] | ||
if caching is CachingEnum.EXPIRES and clean_cache_option: | ||
for match in REGEX_CACHE_UNIT.finditer(clean_cache_option): | ||
plural_unit = match.group("unit").lower() + "s" | ||
number = int(match.group("number")) | ||
logger.info(f"Cache option with unit: {number} {plural_unit}") | ||
expires_after = timedelta(**{plural_unit: number}) | ||
break | ||
else: | ||
logger.warning(f"Invalid cache option: {clean_cache_option}. Defaulting to 1 hour") | ||
|
||
return caching, expires_after |
Oops, something went wrong.