Source code for runway.sources.source

"""Abstract parent class for a 'Source' type object.

Allows us to specify specific remote sourced resources for out application
(Git, S3, ect.)

"""
import logging
import os
from typing import Dict, Optional, Union  # noqa pylint: disable=W

LOGGER = logging.getLogger(__name__)


[docs]class Source(object): """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 in the users home directory: ``~/.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. """ def __init__(self, cache_dir="", **_): # type(Dict[str, Union[str, Dict[str, str]]]) -> Source """Source. Arguments: cache_dir (str): The directory where the given remote resource should be cached """ self.cache_dir = cache_dir if not self.cache_dir: self.cache_dir = os.path.expanduser("~/.runway_cache") # type: str self.__create_cache_directory()
[docs] def fetch(self): # type: () -> None """Retrieve remote source. To be implemented in each subclass.""" raise NotImplementedError
def __create_cache_directory(self): # type: () -> None """If no cache directory exists for the remote runway modules, create one.""" if not os.path.isdir(self.cache_dir): os.mkdir(self.cache_dir)
[docs] @staticmethod def sanitize_directory_path(uri): # type: (str) -> str """Sanitize a Source directory path string. Arguments: uri (str): The uniform resource identifier when targetting a remote resource. """ for i in ["@", "/", ":"]: uri = uri.replace(i, "_") # type: str return uri