62 lines
1.8 KiB
Python
62 lines
1.8 KiB
Python
import logging
|
|
import re
|
|
import subprocess
|
|
from itertools import zip_longest
|
|
from pathlib import Path
|
|
from typing import TYPE_CHECKING, Generator, Sequence
|
|
|
|
import attr
|
|
from ruamel.yaml import CommentedMap
|
|
|
|
from .executable import COMPOSE_EXE
|
|
|
|
if TYPE_CHECKING:
|
|
from .project import Project
|
|
|
|
_logger = logging.getLogger(__name__)
|
|
|
|
|
|
@attr.s
|
|
class Service:
|
|
name: str = attr.ib()
|
|
content: CommentedMap = attr.ib()
|
|
parent: "Project" = attr.ib()
|
|
|
|
_RE_CONFDIR = re.compile(r"^\s*\$(?:CONFDIR|{CONFDIR})/+(.*)$", flags=re.UNICODE)
|
|
|
|
@property
|
|
def configs(self) -> Generator[Path, None, None]:
|
|
if "volumes" not in self.content:
|
|
return
|
|
|
|
for volume in self.content["volumes"]:
|
|
host_part = volume.split(":")[0]
|
|
cd_match = Service._RE_CONFDIR.match(host_part)
|
|
|
|
if cd_match:
|
|
yield Path(cd_match.group(1))
|
|
|
|
def has_executable(self, exe_name: str) -> bool:
|
|
try:
|
|
# test if desired executable exists
|
|
COMPOSE_EXE.run(
|
|
["exec", "-T", self.name, "/bin/sh", "-c", f"command -v {exe_name}"],
|
|
check=True, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
|
|
**self.parent.process_kwargs,
|
|
)
|
|
return True
|
|
|
|
except subprocess.CalledProcessError:
|
|
return False
|
|
|
|
def existing_executables(self, exe_names: Sequence[str]) -> Generator[str, None, None]:
|
|
for cur, nxt in zip_longest(exe_names, exe_names[1:]):
|
|
if self.has_executable(cur):
|
|
# found working shell
|
|
_logger.debug(f"Found executable '{cur}'")
|
|
yield cur
|
|
|
|
elif nxt is not None:
|
|
# try next in list
|
|
_logger.info(f"Executable '{cur}' not found in container, trying '{nxt}'")
|
|
|