All functions were previously in __main__.py but now they've been split out to separate modules for better reuse. Signed-off-by: Alek Ratzloff <alekratz@gmail.com>
51 lines
1.5 KiB
Python
51 lines
1.5 KiB
Python
import functools
|
|
from pathlib import Path
|
|
import time
|
|
from typing import Optional, Union
|
|
|
|
|
|
def file_cache(
|
|
directory: Union[str, Path] = ".",
|
|
format: str = "%Y%m%d%H%M",
|
|
suffix: str = "",
|
|
deduplicate=True,
|
|
):
|
|
def newest_in_dir(path: Union[str, Path]) -> Optional[Path]:
|
|
d = Path(path)
|
|
newest = None
|
|
newest_time = 0.0
|
|
for p in d.glob("*" + suffix):
|
|
if not p.is_file():
|
|
continue
|
|
stats = p.stat()
|
|
if stats.st_mtime > newest_time:
|
|
newest = p
|
|
newest_time = stats.st_mtime
|
|
return newest
|
|
|
|
def decorator(func):
|
|
@functools.wraps(func)
|
|
async def wrapper(*args, **kwargs):
|
|
d = Path(directory)
|
|
d.mkdir(parents=True, exist_ok=True)
|
|
path = Path(directory, time.strftime(format) + suffix)
|
|
if path.exists():
|
|
return path.read_text()
|
|
else:
|
|
text = await func(*args, **kwargs)
|
|
if deduplicate:
|
|
# find the most recent file in the path
|
|
newest_path = newest_in_dir(directory)
|
|
if newest_path:
|
|
newest_text = newest_path.read_text()
|
|
if newest_text == text:
|
|
# Move the old cache to the new path
|
|
newest_path.rename(path)
|
|
return text
|
|
path.write_text(text)
|
|
return text
|
|
|
|
return wrapper
|
|
|
|
return decorator
|