"""Abstract parent class for a 'Source' type object.
Allows us to specify specific remote sourced resources for out application
(Git, S3, ect.)
"""
import logging
from pathlib import Path
from typing import Any, Union
LOGGER = logging.getLogger(__name__)
[docs]class Source:
"""Abstract parent class for a 'Source' type object.
The Source parent class allows us to specify remote resources
for our application via services such as Git or S3. A cache
directory, as part of object's configuration, is automatically
created by default: ``./.runway/cache``.
This folder can be overridden by specifying the ``cache_dir`` property
in the configuration passed.
Every Source type object is expected to have a ``fetch`` method which
will return the folder path at where the module requested resides.
"""
cache_dir: Path
[docs] def __init__(self, *, cache_dir: Union[Path, str], **_: Any):
"""Source.
Args:
cache_dir: The directory where the given remote resource should be
cached.
"""
self.cache_dir = cache_dir if isinstance(cache_dir, Path) else Path(cache_dir)
self.__create_cache_directory()
[docs] def fetch(self) -> Path:
"""Retrieve remote source. To be implemented in each subclass."""
raise NotImplementedError
def __create_cache_directory(self) -> None:
"""If no cache directory exists for the remote runway modules, create one."""
self.cache_dir.mkdir(exist_ok=True, parents=True)
[docs] @staticmethod
def sanitize_directory_path(uri: str) -> str:
"""Sanitize a Source directory path string.
Arguments:
uri: The uniform resource identifier when targeting a remote resource.
"""
for i in ["@", "/", ":"]:
uri = uri.replace(i, "_")
return uri