move to self-managed connections instead of fabric/invoke

README.md

@@ -6,10 +6,15 @@
 
 ![MIT](https://img.shields.io/github/license/carnival-org/carnival)
 
-Software provisioning tool, built on top of [Fabric](http://www.fabfile.org/)
+Software provisioning tool
 
 * Runs on MacOs and Linux
 * Tested on Ubuntu and CentOS
+* Uses mypy strict typing mode [mypy.ini](mypy.ini)
+* Safe, full run chain is validated before run
+
+# Example
+See [carnival_tasks_example.py](carnival_tasks_example.py)
 
 ## Install
 ```bash

carnival/contrib/steps/apt.py

@@ -0,0 +1,238 @@
+import typing
+
+from colorama import Style as S, Fore as F  # type: ignore
+
+from carnival import Step
+from carnival import Connection
+from carnival.steps import validators
+
+
+class GetPackageVersions(Step):
+    """
+    Получить список доступных версий пакета
+    """
+
+    def __init__(self, pkgname: str):
+        self.pkgname = pkgname
+
+    def get_validators(self) -> typing.List[validators.StepValidatorBase]:
+        return [
+            validators.CommandRequiredValidator('apt-cache'),
+        ]
+
+    def run(self, c: Connection) -> typing.List[str]:
+        versions = []
+        result = c.run(f"DEBIAN_FRONTEND=noninteractive apt-cache madison {self.pkgname}", hide=True, warn=True)
+        if result.ok is False:
+            return []
+
+        for line in result.stdout.strip().split("\n"):
+            n, ver, r = line.split("|")
+            versions.append(ver.strip())
+        return versions
+
+
+class GetInstalledPackageVersion(Step):
+    """
+    Получить установленную версию пакета
+
+    :return: Версия пакета если установлен, `None` если пакет не установлен
+    """
+
+    def __init__(self, pkgname: str):
+        self.pkgname = pkgname
+
+    def get_validators(self) -> typing.List[validators.StepValidatorBase]:
+        return [
+            validators.CommandRequiredValidator('dpkg'),
+        ]
+
+    def run(self, c: Connection) -> typing.Optional[str]:
+        """
+        :return: Версия пакета если установлен, `None` если пакет не установлен
+        """
+        result = c.run(
+            f"DEBIAN_FRONTEND=noninteractive dpkg -l {self.pkgname} | grep '{self.pkgname}'",
+            hide=True,
+            warn=True,
+        )
+        if result.ok is False:
+            return None
+
+        installed, pkgn, ver, arch, *desc = result.stdout.strip().split("\n")[0].split()
+        if installed != 'ii':
+            return None
+
+        assert isinstance(ver, str)
+        return ver.strip()
+
+
+class IsPackageInstalled(Step):
+    """
+    Проверить установлен ли пакет
+    Если версия не указана - проверяется любая
+    """
+
+    def __init__(self, pkgname: str, version: typing.Optional[str] = None) -> None:
+        self.pkgname = pkgname
+        self.version = version
+        self.get_installed_package_version = GetInstalledPackageVersion(pkgname=self.pkgname)
+
+    def get_validators(self) -> typing.List[validators.StepValidatorBase]:
+        return self.get_installed_package_version.get_validators()
+
+    def run(self, c: Connection) -> bool:
+        """
+        Проверить установлен ли пакет
+        Если версия не указана - проверяется любая
+        """
+
+        pkgver = self.get_installed_package_version.run(c=c)
+        if self.version is None and pkgver is not None:
+            return True
+
+        if self.version is not None and pkgver == self.version:
+            return True
+
+        return False
+
+
+class ForceInstall(Step):
+    """
+    Установить пакет без проверки установлен ли он
+    """
+
+    def __init__(self, pkgname: str, version: typing.Optional[str] = None, update: bool = False, hide: bool = False):
+        self.pkgname = pkgname
+        self.version = version
+        self.update = update
+        self.hide = hide
+
+    def get_validators(self) -> typing.List[validators.StepValidatorBase]:
+        return [
+            validators.CommandRequiredValidator('apt-get'),
+        ]
+
+    def run(self, c: Connection) -> None:
+        pkgname = self.pkgname
+        if self.version:
+            pkgname = f"{self.pkgname}={self.version}"
+
+        if self.update:
+            c.run("DEBIAN_FRONTEND=noninteractive sudo apt-get update", hide=self.hide)
+
+        c.run(f"DEBIAN_FRONTEND=noninteractive sudo apt-get install -y {pkgname}", hide=self.hide)
+
+
+class Install(Step):
+    """
+    Установить пакет если он еще не установлен в системе
+    """
+    def __init__(
+        self,
+        pkgname: str,
+        version: typing.Optional[str] = None,
+        update: bool = True,
+        hide: bool = False,
+    ) -> None:
+        """
+        :param pkgname: название пакета
+        :param version: версия
+        :param update: запустить apt-get update перед установкой
+        :param hide: скрыть вывод этапов
+        """
+        self.pkgname = pkgname
+        self.version = version
+        self.update = update
+        self.hide = hide
+
+        self.is_package_installed = IsPackageInstalled(pkgname=self.pkgname, version=self.version)
+        self.force_install = ForceInstall(
+            pkgname=self.pkgname, version=self.version,
+            update=self.update, hide=self.hide
+        )
+
+    def get_validators(self) -> typing.List[validators.StepValidatorBase]:
+        return self.is_package_installed.get_validators() + self.force_install.get_validators()
+
+    def run(self, c: Connection) -> bool:
+        """
+        :return: `True` если пакет был установлен, `False` если пакет уже был установлен ранее
+        """
+        if self.is_package_installed.run(c=c):
+            if self.version is not None:
+                installed_version = GetInstalledPackageVersion(self.pkgname).run(c)
+                if installed_version is not None and self.version == installed_version:
+                    return False
+            else:
+                return False
+            return False
+
+        ForceInstall(pkgname=self.pkgname, version=self.version, update=self.update, hide=self.hide).run(c=c)
+        print(f"{S.BRIGHT}{self.pkgname}{S.RESET_ALL}: {F.YELLOW}installed{F.RESET}")
+        return True
+
+
+class InstallMultiple(Step):
+    """
+    Установить несколько пакетов, если они не установлены
+    """
+
+    def __init__(self, pkg_names: typing.List[str], update: bool = True, hide: bool = False) -> None:
+        """
+        :param pkg_names: список пакетов которые нужно установить
+        :param update: запустить apt-get update перед установкой
+        :param hide: скрыть вывод этапов
+        """
+
+        self.pkg_names = pkg_names
+        self.update = update
+        self.hide = hide
+
+    def get_validators(self) -> typing.List[validators.StepValidatorBase]:
+        return [
+            validators.CommandRequiredValidator('apt-get'),
+        ]
+
+    def run(self, c: Connection) -> bool:
+        """
+        :return: `True` если хотя бы один пакет был установлен, `False` если все пакеты уже были установлен ранее
+        """
+        if all([IsPackageInstalled(x).run(c=c) for x in self.pkg_names]):
+            return False
+
+        if self.update:
+            c.run("DEBIAN_FRONTEND=noninteractive sudo apt-get update", hide=self.hide)
+
+        for pkg in self.pkg_names:
+            Install(pkgname=pkg, update=False, hide=self.hide).run(c=c)
+        return True
+
+
+class Remove(Step):
+    """
+    Удалить пакет
+    """
+
+    def __init__(self, pkg_names: typing.List[str], hide: bool = False) -> None:
+        """
+        :param pkg_names: список пакетов которые нужно удалить
+        :param hide: скрыть вывод этапов
+        """
+        self.pkg_names = pkg_names
+        self.hide = hide
+
+    def get_validators(self) -> typing.List[validators.StepValidatorBase]:
+        return [
+            validators.InlineValidator(
+                if_err_true_fn=lambda c: not self.pkg_names,
+                error_message="'pkg_names' must not be empty",
+            ),
+            validators.CommandRequiredValidator('apt-get'),
+        ]
+
+    def run(self, c: Connection) -> None:
+        c.run(
+            f"DEBIAN_FRONTEND=noninteractive sudo apt-get remove --auto-remove -y {' '.join(self.pkg_names)}",
+            hide=self.hide
+        )

carnival/contrib/steps/cli.py

@@ -0,0 +1,20 @@
+import typing
+
+from carnival import Step, Connection
+
+
+class OpenShell(Step):
+    """
+    Запустить интерактивный шелл в папке
+    """
+
+    def __init__(self, shell_cmd: str = "/bin/bash", cwd: typing.Optional[str] = None) -> None:
+        """
+        :param shell_cmd: команда для запуска шелла
+        :param cwd: папка
+        """
+        self.shell_cmd = shell_cmd
+        self.cwd = cwd
+
+    def run(self, c: "Connection") -> typing.Any:
+        c.run(self.shell_cmd, cwd=self.cwd)

carnival/contrib/steps/docker.py

@@ -0,0 +1,119 @@
+import os
+import typing
+
+from colorama import Fore as F, Style as S  # type: ignore
+
+from carnival import Step
+from carnival import Connection
+from carnival.steps import validators, shortcuts
+
+from carnival.contrib.steps import apt, systemd
+
+
+class CeInstallUbuntu(Step):
+    """
+    Установить docker на ubuntu
+    https://docs.docker.com/engine/install/ubuntu/
+    """
+    def __init__(self, docker_version: typing.Optional[str] = None) -> None:
+        """
+        :param docker_version: версия docker-ce
+        """
+        self.docker_version = docker_version
+
+    def get_validators(self) -> typing.List[validators.StepValidatorBase]:
+        return [
+            validators.CommandRequiredValidator("apt-get"),
+            validators.CommandRequiredValidator("curl"),
+        ]
+
+    def run(self, c: Connection) -> None:
+        pkgname = "docker-ce"
+        if apt.IsPackageInstalled(pkgname=pkgname, version=self.docker_version).run(c=c):
+            print(f"{S.BRIGHT}docker-ce{S.RESET_ALL}: {F.GREEN}already installed{F.RESET}")
+
+        print(f"Installing {pkgname}...")
+        c.run("sudo apt-get update")
+        c.run("sudo apt-get install -y apt-transport-https ca-certificates curl software-properties-common")
+        c.run("curl -fsSL https://download.docker.com/linux/ubuntu/gpg | sudo apt-key add -")
+        c.run('sudo add-apt-repository "deb [arch=amd64] https://download.docker.com/linux/ubuntu $(lsb_release -cs) stable"')  # noqa:E501
+
+        apt.ForceInstall(pkgname=pkgname, version=self.docker_version, update=True, hide=True).run(c=c)
+        print(f"{S.BRIGHT}docker-ce{S.RESET_ALL}: {F.YELLOW}installed{F.RESET}")
+
+
+class ComposeInstall(Step):
+    """
+    Установить docker-compose
+    """
+    def __init__(
+        self,
+        version: str = "1.25.1",
+        dest: str = "/usr/bin/docker-compose",
+    ) -> None:
+        """
+        :param version: версия compose
+        :param dest: папка для установки, позразумевается что она должна быт в $PATH
+        """
+        self.version = version
+        self.dest = dest
+
+    def get_validators(self) -> typing.List[validators.StepValidatorBase]:
+        return [
+            validators.CommandRequiredValidator("curl"),
+        ]
+
+    def run(self, c: Connection) -> None:
+        if shortcuts.is_cmd_exist(c, "docker-compose"):
+            print(f"{S.BRIGHT}docker-compose{S.RESET_ALL}: {F.GREEN}already installed{F.RESET}")
+            return
+
+        link = f"https://github.com/docker/compose/releases/download/{self.version}/docker-compose-`uname -s`-`uname -m`"  # noqa:501
+        c.run(f"sudo curl -sL {link} -o {self.dest}")
+        c.run(f"sudo chmod a+x {self.dest}")
+        print(f"{S.BRIGHT}docker-compose{S.RESET_ALL}: {F.GREEN}already installed{F.RESET}")
+
+
+class UploadImageFile(Step):
+    """
+    Залить с локаьного диска tar-образ docker на сервер
+    и загрузить в демон командой `docker save image -o image.tar`
+    """
+    def __init__(
+        self,
+        docker_image_path: str,
+        dest_dir: str = '/tmp/',
+        rm_after_load: bool = False,
+        rsync_opts: typing.Optional[typing.Dict[str, typing.Any]] = None,
+    ):
+        """
+        :param docker_image_path: tar-образ docker
+        :param dest_dir: папка куда заливать
+        :param rm_after_load: удалить образ после загрузки
+        """
+        if not dest_dir.endswith("/"):
+            dest_dir += "/"
+
+        self.docker_image_path = docker_image_path
+        self.dest_dir = dest_dir
+        self.rm_after_load = rm_after_load
+        self.rsync_opts = rsync_opts or {}
+
+    def get_name(self) -> str:
+        return f"{super().get_name()}(src={self.docker_image_path}, dst={self.dest_dir})"
+
+    def get_validators(self) -> typing.List[validators.StepValidatorBase]:
+        return [
+            validators.CommandRequiredValidator("systemctl"),
+            validators.CommandRequiredValidator("docker"),
+        ]
+
+    def run(self, c: Connection) -> None:
+        image_file_name = os.path.basename(self.docker_image_path)
+        systemd.Start("docker").run(c=c)
+
+        shortcuts.rsync(c.host, self.docker_image_path, self.dest_dir, **self.rsync_opts)
+        c.run(f"cd {self.dest_dir}; docker load -i {image_file_name}")
+
+        if self.rm_after_load:
+            c.run(f"rm -rf {self.dest_dir}{image_file_name}")