"""Handler for the ``path`` field of a Runway module."""
from __future__ import annotations
import logging
import re
from pathlib import Path
from typing import TYPE_CHECKING, ClassVar
from urllib.parse import parse_qs
from typing_extensions import TypedDict
from ...compat import cached_property
from ...config.components.runway import RunwayModuleDefinition
from ...config.models.runway import RunwayModuleDefinitionModel
from ...sources.git import Git
from ._deploy_environment import DeployEnvironment
if TYPE_CHECKING:
from ...sources.source import Source
LOGGER = logging.getLogger(__name__)
class ModulePathMetadataTypeDef(TypedDict):
"""Type definition for ModulePath.metadata."""
arguments: dict[str, str]
cache_dir: Path
location: str
source: str
uri: str
[docs]
class ModulePath:
"""Handler for the ``path`` field of a Runway module."""
ARGS_REGEX: ClassVar[str] = r"(\?)(?P<args>.*)$"
REMOTE_SOURCE_HANDLERS: ClassVar[dict[str, type[Source]]] = {"git": Git}
SOURCE_REGEX: ClassVar[str] = r"(?P<source>[a-z]+)(\:\:)"
URI_REGEX: ClassVar[str] = r"(?P<uri>[a-z]+://[a-zA-Z0-9\./-]+?(?=//|\?|$))"
[docs]
def __init__(
self,
definition: Path | str | None = None,
*,
cache_dir: Path,
deploy_environment: DeployEnvironment | None = None,
) -> None:
"""Instantiate class.
Args:
definition: Path definition.
cache_dir: Directory to use for caching if needed.
deploy_environment: Current deploy environment object.
"""
self.cache_dir = cache_dir
self.definition = definition or Path.cwd()
self.env = deploy_environment or DeployEnvironment()
@cached_property
def arguments(self) -> dict[str, str]:
"""Remote source arguments."""
if isinstance(self.definition, str):
match = re.match(rf"^.*{self.ARGS_REGEX}", self.definition)
if match:
return {k: ",".join(v) for k, v in parse_qs(match.group("args")).items()}
return {}
@cached_property
def location(self) -> str:
"""Location of the module."""
if isinstance(self.definition, str):
if re.match(r"^(/|//|\.|\./)", self.definition) or "::" not in self.definition:
return re.sub(self.ARGS_REGEX, "", self.definition)
no_src = re.sub(rf"^{self.SOURCE_REGEX}", "", self.definition)
no_uri = re.sub(rf"^{self.URI_REGEX}", "", no_src)
match = re.search(r"//(?P<location>[^\?\n]*)", no_uri)
if match:
return match.group("location")
return "./"
@cached_property
def metadata(self) -> ModulePathMetadataTypeDef:
"""Information that describes the module path."""
return {
"arguments": self.arguments,
"cache_dir": self.cache_dir,
"location": self.location,
"source": self.source,
"uri": self.uri,
}
@cached_property
def module_root(self) -> Path:
"""Root directory of the module."""
if isinstance(self.definition, Path):
return self.definition
if self.source != "local":
return self._fetch_remote_source()
return self.env.root_dir / self.location
@cached_property
def source(self) -> str:
"""Source of the module."""
if isinstance(self.definition, str):
match = re.match(rf"^{self.SOURCE_REGEX}.*$", self.definition)
if match:
return match.group("source")
return "local"
@cached_property
def uri(self) -> str:
"""Remote source URI."""
if isinstance(self.definition, str):
match = re.match(rf"^{self.SOURCE_REGEX}{self.URI_REGEX}", self.definition)
if match:
return match.group("uri")
return ""
def _fetch_remote_source(self) -> Path:
"""Fetch remote module source.
Raises:
NotImplementedError: The supplied source does not have a handler.
"""
try:
return self.REMOTE_SOURCE_HANDLERS[self.source](**self.metadata).fetch()
except KeyError:
raise NotImplementedError(
f"{self.source} is not a supported Runway module source"
) from None
[docs]
@classmethod
def parse_obj(
cls,
obj: Path | RunwayModuleDefinition | RunwayModuleDefinitionModel | str | None,
*,
cache_dir: Path,
deploy_environment: DeployEnvironment | None = None,
) -> ModulePath:
"""Parse object.
Args:
obj: Object to parse.
cache_dir: Directory to use for caching if needed.
deploy_environment: Current deploy environment object.
Raises:
TypeError: Unsupported type provided.
"""
if isinstance(obj, (RunwayModuleDefinition, RunwayModuleDefinitionModel)):
return cls(
cache_dir=cache_dir,
definition=obj.path,
deploy_environment=deploy_environment,
)
if isinstance(obj, (type(None), Path, str)): # pyright: ignore[reportUnnecessaryIsInstance]
return cls(
cache_dir=cache_dir,
definition=obj,
deploy_environment=deploy_environment,
)
raise TypeError(
f"object type {type(obj)}; expected pathlib.Path, "
"RunwayModuleDefinition, RunwayModuleDefinitionModel, or str"
)