full working POC

Dockerfile

@@ -0,0 +1,26 @@
+FROM    python:3.12-slim
+
+# SSH
+RUN     apt-get update && apt-get -y install openssh-server
+RUN     mkdir -p /var/run/sshd
+
+# RUN     echo StrictHostKeyChecking no >> /etc/ssh/ssh_config
+# RUN     ssh-keygen -t rsa -b 4096 -f ssh_host_rsa_key
+RUN     /usr/bin/ssh-keygen -A
+RUN     mkdir -p /var/run/sshd
+RUN     service ssh restart
+
+# change password root
+RUN     useradd --create-home --shell /bin/bash  shinylive
+RUN     echo "shinylive:docker" | chpasswd
+USER    shinylive
+WORKDIR /home/shinylive
+RUN     mkdir /home/shinylive/shinyapps
+
+USER root
+EXPOSE  5000
+EXPOSE  22
+
+COPY entrypoint.sh .
+
+CMD bash ./entrypoint.sh

entrypoint.sh

@@ -0,0 +1,17 @@
+#!/bin/bash
+cat /etc/ssh/ssh_config
+
+# Start the first process
+echo "Starting SSH and python webserver..."
+# service ssh restart &
+/usr/sbin/sshd &
+
+# Start the second process
+# echo "Starting webserver..."
+python3 -m http.server --directory shinyapps 5000
+
+# Wait for any process to exit
+wait -n
+
+# Exit with status of process that exited first
+exit $?

pyproject.toml

@@ -26,22 +26,6 @@ tests = [
 Homepage = "https://github.com/darrida/py-shinylive-deploy-webserver"
 Issues = "https://github.com/darrida/py-shinylive-deploy-webserver/issues"
 
-[tool.shinylive_deploy]
-app_name = "app1"
-dev_dir = "src_dev/app1"
-deploy_branch = "main"
-beta_branch = "main"
-host = "testserver"
-user = "testuser"
-
-[tool.shinylive_deploy.local]
-deploy_dir = "src_test_webserver/shinyapps/"
-base_url = "localhost:8000/apps"
-
-[tool.shinylive_deploy.webserver]
-deploy_dir = "/homes/user/docker_volumes/shinyapps/"
-base_url = "https://realurl.com/apps"
-
 [build-system]
 requires = ["setuptools>=61.0"]
 build-backend = "setuptools.build_meta"

shiny_deploy.toml

@@ -2,7 +2,7 @@
 app_name = "app1"
 
 [development]
-directory = "src_dev"
+directory = "src_dev/app1"
 
 [deploy.gitbranch]
 prod = "main"
@@ -16,7 +16,8 @@ directory = "src_test_webserver/shinyapps/"
 base_url = "localhost:8000/apps"
 
 [deploy.server]
-host = "testserver"
-user = "testuser"
-directory = "/homes/user/docker_volumes/shinyapps/"
-base_url = "https://realurl.com/apps"
+host = "127.0.0.1"
+user = "shinylive"
+port = 2222
+directory = "shinyapps"
+base_url = "http://localhost:5000/"

shinylive_deploy/main.py

@@ -5,25 +5,40 @@
 from pydantic import SecretStr
 
 if __name__ == "__main__":
-    if input("Server deployment (y/n)? ")  == "y":
-        config = toml["deploy"]["server"]
+    # parse arguments
+    if len(sys.argv) < 2 or sys.argv[1] not in ("test", "beta", "prod", "local"):
+            raise ValueError("\nERROR: One of the following arguments is required -> [ local | test | beta | prod ]\n")
+    deploy_mode = sys.argv[1]
+    try:
+        rollback = sys.argv[2]
+        if rollback not in ("-r", "--rollback"):
+            raise ValueError("2nd optional argument must be `-r` or `--rollback`")
+        rollback = True
+    except IndexError:
+        rollback = False
 
+    # initialize configuration
+    if deploy_mode in ("test", "beta", "prod"):
+        config = toml["deploy"]["server"]
         shinylive_ = ServerShinyDeploy(
+            mode=deploy_mode,
             base_url=config["base_url"],
             dir_deployment=config["directory"],
             host=config["host"],
             user=config["user"],
+            port=config.get("port", 22),
             password=SecretStr(value=getpass(f"SSH password for [{config["user"]}]: "))
         )
-    else:
+    else:  # local
         config = toml["deploy"]["local"]
-
         shinylive_ = LocalShinyDeploy(
+            mode=deploy_mode,
             base_url=config["base_url"],
             dir_deployment=config["directory"]
         )
 
-    if "rollback" in sys.argv:
+    # execute deployment change
+    if rollback is True:
         shinylive_.rollback()
     else:
         shinylive_.deploy()

shinylive_deploy/models.py

@@ -1,27 +1,31 @@
 import subprocess
 import sys
+from contextlib import contextmanager
 from dataclasses import dataclass
 from pathlib import Path, PurePosixPath
 from typing import Literal
 
 import git
+import paramiko
 import tomllib
+from paramiko import AutoAddPolicy, SFTPClient, SSHClient
 from pydantic import SecretStr
 
 repo = git.Repo()
 
 with open("shiny_deploy.toml", "rb") as f:
     toml = tomllib.load(f)
+
 deploy_local = toml["deploy"]["local"]
 deploy_server = toml["deploy"]["server"]
-development: dict = toml["deploy"].get("development", {})
+development: dict = toml.get("development", {})
 staging: dict = toml["deploy"].get("staging", {})
 gitbranch: dict = toml["deploy"].get("gitbranch", {})
 
-
-class DeployException(Exception):
-    pass
-
+subprocess_config = {
+    "capture_output": True, "text": True, 
+    "shell": True, "check": True
+}
 
 @dataclass
 class ShinyDeploy:
@@ -39,72 +43,144 @@ def deploy_name(self):
         modes = {"prod": "", "beta": "-beta", "test": "-test", "local": ""}
         return self.app_name + modes[self.mode]
 
-    def __message(self):
-        modes = {"prod": "PRODUCTION", "beta": "BETA", "test": "TEST", "local": "LOCAL"}
+    def _message(self):
         print(
             "\n##################################"
-            f"\nDEPLOYMENT MODE:   {modes[self.mode]}"
+            f"\nDEPLOYMENT MODE: {self.mode}"
             f"\nDEPLOYMENT NAME: {self.deploy_name}"
             "\n##################################"
         )
 
-    def __check_requirements(self):
-        if len(sys.argv) < 2 or sys.argv[1] not in ("test", "beta", "prod", "local"):
-            raise ValueError("\nERROR: One of the following arguments is required -> [ local | test | beta | prod ]\n")
-        self.mode = sys.argv[1]
+    def _check_requirements(self):
+        # if len(sys.argv) < 2 or sys.argv[1] not in ("test", "beta", "prod", "local"):
+        #     raise ValueError("\nERROR: One of the following arguments is required -> [ local | test | beta | prod ]\n")
+        # self.mode = sys.argv[1]
         if self.mode == "prod" and str(repo.active_branch) != self.prod_branch:
-            raise DeployException(f"Missing Requirement: `prod` deployments can only be executed from the `{self.branches.deploy}` branch")
+            raise DeployException(f"Missing Requirement: `prod` deployments can only be executed from the `{self.prod_branch}` branch")
         elif self.mode == "beta" and str(repo.active_branch) != self.beta_branch:
-            raise DeployException(f"Missing Requirement: `beta` deployments can only be executed from the `{self.branches.beta}` branch")
+            raise DeployException(f"Missing Requirement: `beta` deployments can only be executed from the `{self.beta_branch}` branch")
 
-    def __compile(self):
+    def _compile(self):
         cmd = f"shinylive export {Path(self.dir_development)} {Path(self.dir_staging) / self.deploy_name}"
         print(f"\nExport Command: {cmd}")
         subprocess.run(cmd, shell=True, check=True)
 
-
+        
 @dataclass
 class ServerShinyDeploy(ShinyDeploy):
-    host: str
-    user: str
-    password: SecretStr
+    host: str = None
+    user: str = None
+    port: int = 22
+    password: SecretStr = None
+
+    @property
+    def base_ssh_cmd(self):
+        return f"ssh -p {self.port} {self.user}:{self.password.get_secret_value()}@{self.host}"
+
+    def ssh_connection(self, client: SSHClient) -> SSHClient:
+        client.set_missing_host_key_policy(AutoAddPolicy())
+        client.connect(
+            hostname=self.host, port=self.port, username=self.user, password=self.password.get_secret_value()
+        )
+        return client
+
+    def deployed_dir_exists(self, sftp: SFTPClient):
+        existing_deploy_dir = Path(self.dir_deployment) / self.deploy_name
+        # result = subprocess.run(f"{self.base_ssh_cmd} ls {self.dir_deployment}", **subprocess_config)
+        # directories = result.stdout.split("\n")
+        directories = sftp.listdir(str(self.dir_deployment))
+        if Path(existing_deploy_dir).name in directories:
+            return True
+        return False
+
+    def backup_dir_exists(self, sftp: SFTPClient):
+        existing_deploy_dir = Path(self.dir_deployment) / self.deploy_name
+        # result = subprocess.run(f"{self.base_ssh_cmd} ls {self.dir_deployment}", **subprocess_config)
+        # directories = result.stdout.split("\n")
+        directories = sftp.listdir(str(self.dir_deployment))
+        if Path(f"{existing_deploy_dir}-backup").name in directories:
+            return True
+        return False
 
     def deploy(self, testing: bool = False):
-        self.__check_requirements()
-        self.__message()
-        self.__compile()
-
-        staging_dir = Path(self.dir_staging) / self.deploy_name
-        deployment_dir = PurePosixPath(self.dir_deployment)
-        # if exists:
-        #     cmd = f"ssh {self.user}@{self.host} mv {target_dir} {target_dir}-backup"
-        #     subprocess.run(cmd, shell=True, check=True)
-        cmd = f"pscp -r -i {staging_dir}/ {self.user}@{self.host}:{deployment_dir}"  # /homes/user/docker_volumes/shinyapps/
-        print(f"PSCP Command: {cmd}")
-        if testing:
-            return
-        subprocess.run(cmd, shell=True, check=True)
+        if not all([self.host, self.user, self.password]):
+            raise ValueError("For ServerShinyDeploy, all of the following are required: host, user, password")
+        self._check_requirements()
+        self._message()
+        self._compile()
+
+        with SSHClient() as ssh:
+            ssh = self.ssh_connection(ssh)
+            sftp = ssh.open_sftp()
+
+            staging_dir = Path(self.dir_staging) / self.deploy_name
+            deployment_dir = PurePosixPath(self.dir_deployment) / self.deploy_name
+            if self.deployed_dir_exists(sftp):
+                if self.backup_dir_exists(sftp):
+                    print("\n>>> WARNING <<<: Deployment STOPPED. Backup directory already exists. Delete backup directory, or rollback before redeploying.\n")
+                    return
+                # subprocess.run(f"{self.base_ssh_cmd} mv {deployment_dir} {deployment_dir}-backup", shell=True)
+                sftp.rename(str(deployment_dir), f"{deployment_dir}-backup")
+            # subprocess.run(f"{self.base_ssh_cmd} mkdir shinyapps/app1-test", shell=True, check=True)
+            sftp.mkdir(str(deployment_dir))
+            cmd = f"pscp -P {self.port} -r -pw <PASSWORD> {staging_dir} {self.user}@{self.host}:{self.dir_deployment}"  # /homes/user/docker_volumes/shinyapps/
+            print(f"PSCP Command: {cmd}")
+            if testing:
+                return
+            subprocess.run(cmd.replace("<PASSWORD>", self.password.get_secret_value()), shell=True, check=True)
         print(
             "\nCOMPLETE:"
             f"\n- `{self.app_name}` compiled and deployed to webserver"
             f"\n- App available at {self.base_url}/{self.deploy_name}"
         )
 
     def rollback(self):
-        deployment_dir = PurePosixPath(self.dir_deployment)
-        subprocess.run(f"ssh {self.user}@{self.host} rm -rf {deployment_dir}", shell=True, check=True)
-        subprocess.run(f"ssh {self.user}@{self.host} mv {deployment_dir}-backup {deployment_dir}", shell=True, check=True)
+        self._check_requirements()
+        deployment_dir = PurePosixPath(self.dir_deployment) / self.deploy_name
+
+        with SSHClient() as ssh:
+            ssh = self.ssh_connection(ssh)
+            sftp = ssh.open_sftp()
+            if not self.deployed_dir_exists(sftp):
+                print("\n>>> WARNING <<<: Backback STOPPED. No app directory exists to rollback from.\n")
+                return
+            if not self.backup_dir_exists(sftp):
+                print("\n>>> WARNING <<<: Backback STOPPED. No backup directory exists for rollback.\n")
+                return
+            ssh.exec_command(f"rm -rf {deployment_dir}")
+            ssh.exec_command(f"mv {deployment_dir}-backup {deployment_dir}")
 
 
 class LocalShinyDeploy(ShinyDeploy):
+    def deployed_dir_exists(self):
+        existing_deploy_dir = Path(self.dir_deployment) / self.deploy_name
+        result = subprocess.run(f"ls {self.dir_deployment}", **subprocess_config)
+        directories = result.stdout.split("\n")
+        if Path(existing_deploy_dir).name in directories:
+            return True
+        return False
+
+    def backup_dir_exists(self):
+        existing_deploy_dir = Path(self.dir_deployment) / self.deploy_name
+        result = subprocess.run(f"ls {self.dir_deployment}", **subprocess_config)
+        directories = result.stdout.split("\n")
+        if Path(f"{existing_deploy_dir}-backup").name in directories:
+            return True
+        return False
+
     def deploy(self):
-        self.__check_requirements()
-        self.__message()
-        self.__compile()
+        self._check_requirements()
+        self._message()
+        self._compile()
 
         staging_dir = Path(self.dir_staging) / self.deploy_name
         existing_deploy_dir = Path(self.dir_deployment) / self.deploy_name
-        if existing_deploy_dir.exists():
+        if self.deployed_dir_exists():
+            if self.backup_dir_exists():
+                print("\n>>> WARNING <<<: Deployment STOPPED. Backup directory already exists. Delete backup directory, or rollback before redeploying.\n")
+                return
+            cmd = f"mv {existing_deploy_dir} {existing_deploy_dir}-backup"
+            print(f"Backup move command: {cmd}")
             subprocess.run(f"mv {existing_deploy_dir} {existing_deploy_dir}-backup", shell=True)
         cmd = f"mv {staging_dir} {Path(self.dir_deployment)}"
         print(f"Local Move Command: {cmd}")
@@ -116,6 +192,17 @@ def deploy(self):
         )
 
     def rollback(self):
+        self._check_requirements()
         existing_deploy_dir = Path(self.dir_deployment) / self.deploy_name
-        subprocess.run(f"rm -r {existing_deploy_dir}", shell=True, check=True)
-        subprocess.run(f"mv {existing_deploy_dir}-backup {existing_deploy_dir}", shell=True, check=True)
+        if not self.deployed_dir_exists():
+            print("\n>>> WARNING <<<: Backback STOPPED. No app directory exists to rollback from.\n")
+            return
+        if not self.backup_dir_exists():
+            print("\n>>> WARNING <<<: Backback STOPPED. No backup directory exists for rollback.\n")
+            return
+        subprocess.run(f"rm -r {existing_deploy_dir}", **subprocess_config)
+        subprocess.run(f"mv {existing_deploy_dir}-backup {existing_deploy_dir}", **subprocess_config)
+
+
+class DeployException(Exception):
+    pass