Auto-warp commands in sh -c to support pipes

.vscode/settings.json

@@ -0,0 +1,6 @@
+{
+    "[python]": {
+        "editor.defaultFormatter": "ms-python.black-formatter"
+    },
+    "python.formatting.provider": "none"
+}

up_cli/demos/10_simple_prompt.sh

@@ -2,3 +2,5 @@
 alias up="poetry run up"
 
 up ansible --version
+
+up ansible --version | head -n1 | awk '{print $2}'

up_cli/pyproject.toml

@@ -14,6 +14,7 @@ pluggy = "^1.3.0"
 docker = "^6.1.3"
 up_ansible = { path = "../up_ansible", develop = true }
 up_splat = { path = "../up_splat", develop = true }
+dynaconf = {extras = ["all"], version = "^3.2.3"}
 
 
 [build-system]

up_cli/src/up_cli/__init__.py

@@ -1,8 +1,10 @@
 import pluggy
 from typing import TypeAlias
 
+Context:TypeAlias = dict[str, str]
 Prompt:TypeAlias = list[str]
 
+
 # Is this the right way to do this?
 from .containers import RunConfig
 from .match import match_prompt

up_cli/src/up_cli/containers.py

@@ -1,5 +1,6 @@
 import logging as log
 import docker
+import subprocess
 from dataclasses import dataclass, field
 from typing import TypeAlias
 
@@ -23,13 +24,22 @@ def run(self, run: RunConfig):
         log.debug("Running container: %s", run)
         client = docker.from_env()
         #TODO: Catch errors, print properly, pass all params
-        result = client.containers.run(
-            image=run.image, 
-            command=run.command,
-            auto_remove=run.auto_remove)        
-        log.info("container result")
-        log.info("%s", result)
-        log.debug("Container run done")
+        #TODO: Locate bash properly
+        #TODO: Consider if every command should be auto-wrapped in bash (perhaops detect if contains pipes or redirects)
+        command = ["sh", "-c", subprocess.list2cmdline(run.command)]
+        log.debug("$: %s", run)
+
+        try:
+            result = client.containers.run(
+                image=run.image, 
+                command= command,
+                auto_remove=run.auto_remove)
+            log.debug("container result: \n %s", result)
+
+        except Exception as e:
+            log.debug("Failed to run container")
+            log.debug("%s", run)
+            log.error("%s", e)
 
 class Containers:
     delegate = DockerContainers()

up_cli/src/up_cli/main.py

@@ -1,26 +1,31 @@
 import sys
 import logging as log
 from datetime import datetime
+import shlex
 
-from up_cli import pm, containers
+from up_cli import *
 from .plugins import load_plugins
 from .containers import *
 
 containers = Containers()
 
+
 def print_help():
     log.debug("Please please help me")
 
+
 def exit_cli(code=-1):
     if code:
         log.error("Exiting up cli with code %s", code)
     sys.exit(code)
 
+
 def init():
     sys.set_int_max_str_digits(999999)
     level = log.DEBUG
     log.basicConfig(stream=sys.stderr, level=level)
 
+
 def main():
     init()
     now = datetime.now()
@@ -33,25 +38,41 @@ def main():
     executable = args[0]
     prompt = args[1:]
     log.debug(f"executable: {executable}")
-    log.debug(f"prompt: {prompt}")
-    context = {
-        "executable": executable
-    }
+    context = {"executable": executable}
+    up(context, prompt)
+
+
+def up(context: Context, prompt: Prompt): 
     if not prompt:
         log.error("No prompts found")
         exit_cli("NO_PROMPT_FOUND")
+    if len(prompt) == 1 and " " in prompt[0]:
+        prompt = shlex.split(prompt[0])
+    log.debug(f"prompt: {prompt}")
     load_plugins(context)
-    run_configs = to_run_configs(prompt)
-    log.debug("run_configs: %s", run_configs)  
+    run_configs = run_configs_for_prompt(prompt)
+    log.debug("run_configs: %s", run_configs)
     for run_config in run_configs:
         containers.run(run_config)
 
-def to_run_configs(prompt:list[str]) -> list[RunConfig]:
+
+def run_configs_for_prompt(prompt) -> list[RunConfig]:
+    from_plugins = run_configs_from_plugins(prompt)
+    from_configs = run_configs_from_dynaconf(prompt)
+    result = from_plugins + from_configs
+    return result
+
+
+def run_configs_from_dynaconf(prompt: list[str]) -> list[RunConfig]:
+    return []
+
+
+def run_configs_from_plugins(prompt: list[str]) -> list[RunConfig]:
     result = pm.hook.run_for_prompt(prompt=prompt)
     if not result:
         result = []
-    return result    
+    return result
 
 
-if __name__ == '__main__':
+if __name__ == "__main__":
     main()