default to use parent stdin

Author: jet-logic

.github/workflows/publish.yml

@@ -1,4 +1,4 @@
-name: Build and Publish to PyPI
+name: Publish
 
 on:
   push:

pyproject.toml

@@ -5,6 +5,12 @@ build-backend = "setuptools.build_meta"
 [tool.setuptools.dynamic]
 version = {attr = "runce.__init__.__version__"}
 
+
+[tool.black]
+line-length = 127
+
+[tool.flake8]
+max-line-length = 127
 [project]
 name = "runce"
 dynamic = ["version"]
@@ -17,10 +23,6 @@ classifiers = [
     "Development Status :: 4 - Beta",
     "Intended Audience :: Developers",
     "Programming Language :: Python :: 3",
-    "Programming Language :: Python :: 3.7",
-    "Programming Language :: Python :: 3.8",
-    "Programming Language :: Python :: 3.9",
-    "Programming Language :: Python :: 3.10",
 ]
 
 [project.scripts]

runce/cli.py

@@ -1,11 +1,9 @@
 from argparse import ArgumentParser
 from shlex import join
-from sys import stderr, stdout
-from shutil import copyfileobj
+from sys import stderr, stdout, platform
 from subprocess import Popen, PIPE, run
-import sys
-from typing import Dict, Any
-from .utils import check_pid, filesizepu, kill_pid, tail_bytes, tail_file
+from typing import Any
+from .utils import check_pid, filesizepu, kill_pid, tail_bytes
 from .main import Main, flag, arg
 from .procdb import ProcessDB as Manager
 
@@ -25,13 +23,13 @@ def __missing__(self, key: str) -> str:
                 return self["cmd"]
             return join(self["cmd"])
         elif key == "pid_status":
-            return "Running" if check_pid(self["pid"]) else "Stopped"
+            return "Live" if check_pid(self["pid"]) else "Done"
         raise KeyError(f"No {key!r}")
 
 
 def format_prep(f: str):
 
-    def fn(x: Dict[str, Any]) -> str:
+    def fn(x: "dict[str, Any]") -> str:
         return f.format_map(FormatDict(x))
 
     return fn
@@ -56,7 +54,7 @@ def ambiguous(name):
 class Clean(Main):
     """Clean up dead processes."""
 
-    ids: list[str] = arg("ID", "run ids", nargs="*")
+    ids: "list[str]" = arg("ID", "run ids", nargs="*")
 
     def add_arguments(self, argp: ArgumentParser) -> None:
         argp.description = "Clean up entries for non-existing processes"
@@ -78,7 +76,7 @@ def start(self) -> None:
 class Status(Main):
     """Check process status."""
 
-    ids: list[str] = arg("ID", "run ids", nargs="*")
+    ids: "list[str]" = arg("ID", "run ids", nargs="*")
     format: str = flag(
         "f",
         "format of entry line",
@@ -99,7 +97,7 @@ def start(self) -> None:
 class Kill(Main):
     """Kill running processes."""
 
-    ids: list[str] = arg("ID", "run ids", nargs="+")
+    ids: "list[str]" = arg("ID", "run ids", nargs="+")
     dry_run: bool = flag("dry-run", "dry run (don't actually kill)", default=False)
     remove: bool = flag("remove", "remove entry after killing", default=False)
     group: bool = flag("group", "kill process group", default=False)
@@ -139,7 +137,7 @@ def _tail(n: float, u="", out="", tab=None):
     if u:
         stdout.buffer.write(tail_bytes(out, int(n)))
     elif n > 0:
-        if sys.platform.startswith("win"):
+        if platform.startswith("win"):
             cmd = [
                 "powershell",
                 "-c",
@@ -158,19 +156,16 @@ def _tail(n: float, u="", out="", tab=None):
 class Tail(Main):
     """Tail process output."""
 
-    ids: list[str] = arg("ID", "run ids", nargs="*")
+    ids: "list[str]" = arg("ID", "run ids", nargs="*")
     format: str = flag("header", "header format")
     lines: str = flag("n", "lines", "how many lines or bytes")
-    existing: bool = flag(
-        "x", "only-existing", "only show existing processes", default=False
-    )
+    existing: bool = flag("x", "only-existing", "only show existing processes", default=False)
     tab: bool = flag("t", "tab", "prefix tab space", default=False)
     err: bool = flag("e", "err", "output stderr", default=False)
     p_open: str = "=== "
     p_close: str = " ==="
 
     def start(self) -> None:
-        import sys
 
         if self.format == "no":
             hf = None
@@ -194,7 +189,7 @@ def start(self) -> None:
 class Run(Main):
     """Run a new singleton process."""
 
-    args: list[str] = arg("ARG", nargs="*", metavar="arg")
+    args: "list[str]" = arg("ARG", nargs="*", metavar="arg")
     tail: int = flag("t", "tail", "tail the output with n lines", default=0)
     run_id: str = flag("id", "Unique run identifier", default="")
     cwd: str = flag("Working directory for the command")
@@ -219,9 +214,7 @@ def start(self) -> None:
             s = ["🚨", r"Found: PID={pid} ({pid_status}) {name}"]
         else:
             # Start new process
-            e = sp.spawn(
-                args, name, overwrite=self.overwrite, cwd=self.cwd, split=self.split
-            )
+            e = sp.spawn(args, name, overwrite=self.overwrite, cwd=self.cwd, split=self.split)
             s = ["🚀", r"Started: PID={pid} ({pid_status}) {name}"]
         assert e
         try:
@@ -271,7 +264,7 @@ def start(self) -> None:
 class Restart(Main):
     """Restart a process."""
 
-    ids: list[str] = arg("ID", "run ids", nargs="+")
+    ids: "list[str]" = arg("ID", "run ids", nargs="+")
     tail: int = flag("t", "tail", "tail the output with n lines", default=0)
 
     def init_argparse(self, argp: ArgumentParser) -> None:
@@ -294,8 +287,7 @@ class App(Main):
     def init_argparse(self, argp: ArgumentParser) -> None:
         argp.prog = "runce"
         argp.description = (
-            "Runce (Run Once) - Ensures commands run exactly once.\n"
-            "Guarantees singleton execution per unique ID."
+            "Runce (Run Once) - Ensures commands run exactly once.\n" "Guarantees singleton execution per unique ID."
         )
         return super().init_argparse(argp)
 

runce/main.py

@@ -1,6 +1,6 @@
 from typing import TYPE_CHECKING
 
-__version__ = "0.0.6"
+__version__ = "0.0.7"
 if TYPE_CHECKING:
     from argparse import ArgumentParser
     from typing import Sequence, Generator
@@ -15,28 +15,20 @@ def __init__(self, *args: str, **kwargs):
         self.args = args
         self.kwargs = kwargs
 
-    def _add(
-        self, name: str, type_: type, argp: "ArgumentParser", that: object
-    ) -> None:
+    def _add(self, name: str, klass: type, argp: "ArgumentParser", that: object) -> None:
         """Add argument to parser."""
         args = []
         kwargs = {**self.kwargs}
         flag_arg = kwargs.pop("flag", None)
         action = kwargs.get("action")
         const = kwargs.get("const")
         default = kwargs.get("default", INVALID)
+        kind = klass if isinstance(klass, type) else type
 
         if action is None:
-
             if const is not None:
-                kwargs["action"] = (
-                    "append_const"
-                    if issubclass(type_, list) or isinstance(default, list)
-                    else "store_const"
-                )
-            elif type_ is None:
-                kwargs["action"] = "store"
-            elif isinstance(type_, type) and issubclass(type_, bool):
+                kwargs["action"] = "append_const" if issubclass(kind, list) or isinstance(default, list) else "store_const"
+            elif issubclass(kind, bool):
                 if default is None:
                     try:
                         from argparse import BooleanOptionalAction
@@ -49,9 +41,7 @@ def _add(
                 else:
                     assert default is INVALID or default is False
                     kwargs["action"] = "store_true"
-            elif (isinstance(type_, type) and issubclass(type_, list)) or isinstance(
-                default, list
-            ):
+            elif issubclass(kind, list) or isinstance(default, list):
                 if "nargs" not in kwargs:
                     kwargs["action"] = "append"
                 if "default" not in kwargs:
@@ -60,17 +50,13 @@ def _add(
                 kwargs["action"] = "store"
 
         parser = kwargs.pop("parser", None)
-        if kwargs.get("action") == "count":
+        if kwargs.get("action") in ("count", "store_const"):
             pass
         elif parser:
             kwargs["type"] = parser
-        elif (
-            type_ is not bool
-            and type(type_) is type
-            and issubclass(type_, (int, float, str))
-        ):
-            kwargs["type"] = type_
-        # print(name, type_, that, "_add", action, flag_arg)
+        elif kind is not bool and issubclass(kind, (int, float, str)):
+            kwargs["type"] = kind
+
         if flag_arg is None:
             for x in self.args:
                 if " " in x or "\t" in x:
@@ -83,9 +69,7 @@ def _add(
         else:
 
             def add_args(x: str) -> None:
-                args.append(
-                    x if x.startswith("-") else (f"--{x}" if len(x) > 1 else f"-{x}")
-                )
+                args.append(x if x.startswith("-") else (f"--{x}" if len(x) > 1 else f"-{x}"))
 
             for x in self.args:
                 if " " in x or "\t" in x:
@@ -135,17 +119,11 @@ def __getattr__(self, name: str) -> object:
         try:
             m = super().__getattr__
         except AttributeError:
-            raise AttributeError(
-                f"{self.__class__.__name__} has no attribute {name}"
-            ) from None
+            raise AttributeError(f"{self.__class__.__name__} has no attribute {name}") from None
         else:
             return m(name)
 
-    def main(
-        self,
-        args: "Sequence[str]|None" = None,
-        argp: "ArgumentParser|None" = None,
-    ):
+    def main(self, args: "Sequence[str]|None" = None, argp: "ArgumentParser|None" = None):
         """Entry point for CLI execution.
         Args:
             args: Command-line arguments (optional).
@@ -190,9 +168,7 @@ def sub_args(self):
         """Yield subcommands."""
         yield None, {}
 
-    def parse_arguments(
-        self, argp: "ArgumentParser", args: "Sequence[str]|None"
-    ) -> None:
+    def parse_arguments(self, argp: "ArgumentParser", args: "Sequence[str]|None") -> None:
         """Parse command line arguments."""
         p = self._walk_subparsers(argp)
 
@@ -226,12 +202,14 @@ def _walk_subparsers(self, argp: "ArgumentParser", root=None):
         return s
 
     def _arg_parents_and_self(self):
+        # type: () -> Generator[Main, object, None]
         c: "Main | None" = self
         while c is not None:
             yield c
             c = c._arg_parent
 
     def _arg_parents(self):
+        # type: () -> Generator[Main, object, None]
         c = self._arg_parent
         while c is not None:
             yield c

runce/spawn.py

@@ -2,7 +2,7 @@
 from json import dump, load
 from uuid import uuid4
 from pathlib import Path
-from subprocess import DEVNULL, STDOUT, Popen
+from subprocess import DEVNULL, PIPE, STDOUT, Popen
 from time import time
 from .utils import generate_pseudowords, get_base_name, look_multiple
 
@@ -23,9 +23,7 @@ def __getattr__(self, name: str) -> object:
         try:
             m = super().__getattr__
         except AttributeError:
-            raise AttributeError(
-                f"{self.__class__.__name__} has no attribute {name}"
-            ) from None
+            raise AttributeError(f"{self.__class__.__name__} has no attribute {name}") from None
         else:
             return m(name)
 
@@ -77,7 +75,10 @@ def spawn(
             so = se = Path(out_file) if out_file else data_dir / f"{base_name}.log"
             po_kwa["stdout"] = so.open(f"{mode}b")
             po_kwa["stderr"] = STDOUT
+        if po_kwa.get("stdin") is None:
+            from sys import stdin
 
+            po_kwa["stdin"] = stdin.buffer
         po_kwa.setdefault("start_new_session", True)
         po_kwa.setdefault("close_fds", True)
         po_kwa.setdefault("stdin", DEVNULL)
@@ -112,11 +113,7 @@ def all(self):
             return
 
         for child in self.data_dir.iterdir():
-            if (
-                child.is_file()
-                and child.name.endswith(".run.json")
-                and child.stat().st_size > 0
-            ):
+            if child.is_file() and child.name.endswith(".run.json") and child.stat().st_size > 0:
                 try:
                     with child.open() as f:
                         d: dict[str, int | str] = load(f)
@@ -143,9 +140,7 @@ def drop(self, entry: "dict[str, object]", clean_up=True):
                 if clean_up or k == "file":
                     remove(v)
 
-    def find_names(
-        self, names: "list[str]", ambiguous=lambda x: None, not_found=lambda x: None
-    ):
+    def find_names(self, names: "list[str]", ambiguous=lambda x: None, not_found=lambda x: None):
         if names:
             yield from look_multiple(names, self.all(), ambiguous, not_found)
         else:

tests/test_cli.py

@@ -21,7 +21,7 @@ def run_runce(self, *args, stdout_only=False):
         # Combine stdout and stderr for verification
         return o
 
-    def assertRegexInListOnce(self, regex: str, ls: list[str]):
+    def assertRegexInListOnce(self, regex: str, ls: 'list[str]'):
         b = [1 for x in ls if re.search(regex, x)]
         self.assertEqual(len(b), 1, f"{regex} in {ls!r}")