Created
May 5, 2025 14:02
-
-
Save hnakamur/6b74f3af7f75999a58a59109328fdef0 to your computer and use it in GitHub Desktop.
a Python script to run a command in a detached tmux session, forwarding this script's stdin to the command if stdin is not terminal.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
#!/usr/bin/env python3 | |
from __future__ import annotations | |
import argparse | |
import logging | |
import os | |
import shlex | |
import subprocess | |
import sys | |
import tempfile | |
from types import TracebackType | |
from typing import Optional, Type | |
PROG_NAME = os.path.basename(__file__) | |
class NamedPipe: | |
def __init__(self, filename_prefix: str) -> None: | |
[_, pipe_filename] = tempfile.mkstemp(prefix=filename_prefix) | |
os.unlink(pipe_filename) | |
os.mkfifo(pipe_filename, mode=0o600) | |
self.name = pipe_filename | |
def __enter__(self) -> NamedPipe: | |
return self | |
def __exit__( | |
self, | |
exctype: Optional[Type[BaseException]], | |
excinst: Optional[BaseException], | |
exctb: Optional[TracebackType], | |
) -> None: | |
os.unlink(self.name) | |
def parse_args() -> argparse.Namespace: | |
parser = argparse.ArgumentParser( | |
prog=PROG_NAME, | |
description="""Run a command in a detached tmux session, | |
forwarding this script's stdin to the command if stdin is not terminal.""", | |
) | |
parser.add_argument( | |
"-s", | |
"--session-name", | |
required=True, | |
help="tmux session name", | |
) | |
parser.add_argument( | |
"--debug", | |
action="store_true", | |
help="Enable debug log", | |
) | |
parser.add_argument( | |
"command", | |
nargs="+", | |
help="command and arguments to execute", | |
) | |
return parser.parse_args() | |
def config_logging(debug: bool) -> None: | |
if debug: | |
logging.basicConfig( | |
level=logging.DEBUG, | |
format="%(asctime)s.%(msecs)03d %(levelname)s %(message)s", | |
datefmt="%Y-%m-%d %H:%M:%S", | |
) | |
else: | |
logging.basicConfig( | |
level=logging.INFO, | |
format="%(message)s", | |
) | |
def run_cmd_detached_on_tmux( | |
session_name: str, command: list[str], forward_stdin: bool = False | |
) -> int: | |
logging.debug( | |
f"run_cmd_detached_on_tmux, session_name={session_name}, command={command}, foward_stdin={forward_stdin}" | |
) | |
if forward_stdin: | |
with NamedPipe(f"{PROG_NAME}-named-pipe.") as pipe: | |
session_cmd = " ".join( | |
["<", pipe.name] + [shlex.quote(term) for term in command] | |
) | |
logging.debug(f"session_cmd={session_cmd}") | |
cmdline = f"tmux new-session -d -s {session_name} -- sh -c {shlex.quote(session_cmd)}" | |
logging.debug(f"cmdline={cmdline}") | |
tmux_proc = subprocess.run(cmdline, shell=True, capture_output=True) | |
returncode = tmux_proc.returncode | |
if returncode != 0: | |
errmsg = tmux_proc.stderr.decode().strip() | |
logging.error( | |
f"exiting since tmux failed with returncode={returncode}, err={errmsg}" | |
) | |
return returncode | |
with open(pipe.name, "w") as f: | |
for line in sys.stdin: | |
f.write(line) | |
logging.debug(f"written to {pipe.name}: {line.strip()}") | |
logging.debug(f"closed {pipe.name}") | |
else: | |
session_cmd = " ".join([shlex.quote(term) for term in command]) | |
logging.debug(f"session_cmd={session_cmd}") | |
cmdline = ( | |
f"tmux new-session -d -s {session_name} -- sh -c {shlex.quote(session_cmd)}" | |
) | |
logging.debug(f"cmdline={cmdline}") | |
tmux_proc = subprocess.run(cmdline, shell=True, capture_output=True) | |
returncode = tmux_proc.returncode | |
if returncode != 0: | |
errmsg = tmux_proc.stderr.decode().strip() | |
logging.error( | |
f"exiting since tmux failed with returncode={returncode}, err={errmsg}" | |
) | |
return returncode | |
return 0 | |
def main(): | |
args = parse_args() | |
config_logging(args.debug) | |
forward_stdin = not sys.stdin.isatty() | |
rc = run_cmd_detached_on_tmux(args.session_name, args.command, forward_stdin) | |
sys.exit(rc) | |
main() |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment