| import os |
| from os import PathLike |
| from pathlib import Path |
| from typing import Tuple, Union |
| from urllib.parse import urlparse |
|
|
| PathOrStr = Union[str, PathLike] |
|
|
| CACHE_DIRECTORY: PathOrStr = Path( |
| os.getenv("CACHED_PATH_CACHE_ROOT", Path.home() / ".cache" / "cached_path") |
| ) |
| """ |
| The default global cache directory. |
| """ |
|
|
|
|
| def _split_cloud_path(url: str, provider: str) -> Tuple[str, str]: |
| """Split a full s3 path into the bucket name and path.""" |
| parsed = urlparse(url) |
| if not parsed.netloc or not parsed.path: |
| raise ValueError("bad {} path {}".format(provider, url)) |
| bucket_name = parsed.netloc |
| provider_path = parsed.path |
| |
| if provider_path.startswith("/"): |
| provider_path = provider_path[1:] |
| return bucket_name, provider_path |
|
|
|
|
| def set_cache_dir(cache_dir: PathOrStr) -> None: |
| """ |
| Set the global default cache directory. |
| """ |
| global CACHE_DIRECTORY |
| CACHE_DIRECTORY = Path(cache_dir) |
|
|
|
|
| def get_cache_dir() -> Path: |
| """ |
| Get the global default cache directory. |
| """ |
| return Path(CACHE_DIRECTORY) |
|
|
