Added FuncAlias

.pre-commit-config.yaml

@@ -8,7 +8,7 @@ default_language_version:
 repos:
   - repo: https://github.com/astral-sh/ruff-pre-commit
     # Ruff version.
-    rev: 'v0.4.2'
+    rev: 'v0.4.3'
     hooks:
       - id: ruff
         args: [., --fix, --exit-non-zero-on-fix]

news/aliases_imp.rst

@@ -0,0 +1,23 @@
+**Added:**
+
+* Added FuncAlias to process callable aliases.
+
+**Changed:**
+
+* <news item>
+
+**Deprecated:**
+
+* <news item>
+
+**Removed:**
+
+* <news item>
+
+**Fixed:**
+
+* <news item>
+
+**Security:**
+
+* <news item>

tests/test_aliases.py

@@ -9,7 +9,7 @@
 from xonsh.aliases import Aliases, ExecAlias
 
 
-def cd(args, stdin=None, **kwargs):
+def cd(args, stdin=None):
     return args
 
 
@@ -30,10 +30,11 @@ def test_imports(xession):
         "o": ["omg", "lala"],
         "ls": ["ls", "-  -"],
         "color_ls": ["ls", "--color=true"],
-        "cd": cd,
+        "cd": "FuncAlias",
         "indirect_cd": ["cd", ".."],
     }
     raw = ales._raw
+    raw["cd"] = type(ales["cd"]).__name__
     assert raw == expected
 
 

xonsh/aliases.py

@@ -40,6 +40,7 @@
     argvquote,
     escape_windows_cmd_string,
     print_color,
+    print_exception,
     strip_simple_quotes,
     swap_values,
     to_repr_pretty_,
@@ -54,6 +55,40 @@ def EXEC_ALIAS_RE():
     return re.compile(r"@\(|\$\(|!\(|\$\[|!\[|\&\&|\|\||\s+and\s+|\s+or\s+|[>|<]")
 
 
+class FuncAlias:
+    """Provides a callable alias for xonsh commands."""
+
+    attributes_inherit = ["__xonsh_threadable__", "__xonsh_capturable__", "__doc__"]
+    attributes_show = ["__xonsh_threadable__", "__xonsh_capturable__"]
+
+    def __init__(self, name, func):
+        self.__name__ = self.name = name
+        self.func = func
+        for attr in self.attributes_inherit:
+            if (val := getattr(func, attr, None)) is not None:
+                self.__setattr__(attr, val)
+
+    def __repr__(self):
+        r = {"name": self.name, "func": self.func}
+        r |= {
+            attr: val
+            for attr in self.attributes_show
+            if (val := getattr(self, attr, None)) is not None
+        }
+        return f"FuncAlias({repr(r)})"
+
+    def __call__(
+        self, args=None, stdin=None, stdout=None, stderr=None, spec=None, stack=None
+    ):
+        try:
+            func_args = [args, stdin, stdout, stderr, spec, stack][
+                : len(inspect.signature(self.func).parameters)
+            ]
+            return self.func(*func_args)
+        except Exception:
+            print_exception(f"Exception in {repr(self)}")
+
+
 class Aliases(cabc.MutableMapping):
     """Represents a location to hold and look up aliases."""
 
@@ -182,6 +217,8 @@ def __setitem__(self, key, val):
             else:
                 # need to exec alias
                 self._raw[key] = ExecAlias(val, filename=f)
+        elif isinstance(val, types.FunctionType):
+            self._raw[key] = FuncAlias(key, val)
         else:
             self._raw[key] = val
 
@@ -225,7 +262,7 @@ def __repr__(self):
 
 
 class ExecAlias:
-    """Provides a callable alias for xonsh source code."""
+    """Provides an exec alias for xonsh source code."""
 
     def __init__(self, src, filename="<exec-alias>"):
         """

xonsh/procs/proxies.py

@@ -9,7 +9,6 @@
 
 import collections.abc as cabc
 import functools
-import inspect
 import io
 import os
 import signal
@@ -274,7 +273,7 @@ def parse_proxy_return(r, stdout, stderr):
             stdout.write(str(r[0]))
             stdout.flush()
         if rlen > 1 and r[1] is not None:
-            stderr.write(str(r[1]))
+            stderr.write(xt.endswith_newline(str(r[1])))
             stderr.flush()
         if rlen > 2 and isinstance(r[2], int):
             cmd_result = r[2]
@@ -285,69 +284,6 @@ def parse_proxy_return(r, stdout, stderr):
     return cmd_result
 
 
-def proxy_zero(f, args, stdin, stdout, stderr, spec, stack):
-    """Calls a proxy function which takes no parameters."""
-    return f()
-
-
-def proxy_one(f, args, stdin, stdout, stderr, spec, stack):
-    """Calls a proxy function which takes one parameter: args"""
-    return f(args)
-
-
-def proxy_two(f, args, stdin, stdout, stderr, spec, stack):
-    """Calls a proxy function which takes two parameter: args and stdin."""
-    return f(args, stdin)
-
-
-def proxy_three(f, args, stdin, stdout, stderr, spec, stack):
-    """Calls a proxy function which takes three parameter: args, stdin, stdout."""
-    return f(args, stdin, stdout)
-
-
-def proxy_four(f, args, stdin, stdout, stderr, spec, stack):
-    """Calls a proxy function which takes four parameter: args, stdin, stdout,
-    and stderr.
-    """
-    return f(args, stdin, stdout, stderr)
-
-
-def proxy_five(f, args, stdin, stdout, stderr, spec, stack):
-    """Calls a proxy function which takes four parameter: args, stdin, stdout,
-    stderr, and spec.
-    """
-    return f(args, stdin, stdout, stderr, spec)
-
-
-PROXIES = (proxy_zero, proxy_one, proxy_two, proxy_three, proxy_four, proxy_five)
-
-
-def partial_proxy(f):
-    """Dispatches the appropriate proxy function based on the number of args."""
-    numargs = 0
-
-    for name, param in inspect.signature(f).parameters.items():
-        # handle *args/**kwargs signature
-        if param.kind in {param.VAR_KEYWORD, param.VAR_POSITIONAL}:
-            numargs = 6
-            break
-        if (
-            param.kind == param.POSITIONAL_ONLY
-            or param.kind == param.POSITIONAL_OR_KEYWORD
-        ):
-            numargs += 1
-        elif name in xt.ALIAS_KWARG_NAMES and param.kind == param.KEYWORD_ONLY:
-            numargs += 1
-    if numargs < 6:
-        return functools.partial(PROXIES[numargs], f)
-    elif numargs == 6:
-        # don't need to partial.
-        return f
-    else:
-        e = "Expected proxy with 6 or fewer arguments for {}, not {}"
-        raise xt.XonshError(e.format(", ".join(xt.ALIAS_KWARG_NAMES), numargs))
-
-
 def get_proc_proxy_name(cls):
     func_name = cls.f
     if type(cls.f) is functools.partial:
@@ -409,8 +345,7 @@ def __init__(
         env : Mapping, optional
             Environment mapping.
         """
-        self.orig_f = f
-        self.f = partial_proxy(f)
+        self.f = f
         self.args = args
         self.pid = None
         self.returncode = None
@@ -799,8 +734,7 @@ def __init__(
         close_fds=False,
         env=None,
     ):
-        self.orig_f = f
-        self.f = partial_proxy(f)
+        self.f = f
         self.args = args
         self.pid = os.getpid()
         self.returncode = None

xonsh/procs/specs.py

@@ -488,10 +488,18 @@ def _run_binary(self, kwargs):
         except FileNotFoundError as ex:
             cmd0 = self.cmd[0]
             if len(self.cmd) == 1 and cmd0.endswith("?"):
-                with contextlib.suppress(OSError):
-                    return self.cls(
-                        ["man", cmd0.rstrip("?")], bufsize=bufsize, **kwargs
+                cmdq = cmd0.rstrip("?")
+                if cmdq in XSH.aliases:
+                    alias = XSH.aliases[cmdq]
+                    descr = (
+                        repr(alias) + (":\n" + doc)
+                        if (doc := getattr(alias, "__doc__", ""))
+                        else ""
                     )
+                    return self.cls(["echo", descr], bufsize=bufsize, **kwargs)
+                else:
+                    with contextlib.suppress(OSError):
+                        return self.cls(["man", cmdq], bufsize=bufsize, **kwargs)
             e = f"xonsh: subprocess mode: command not found: {repr(cmd0)}"
             env = XSH.env
             sug = xt.suggest_commands(cmd0, env)
@@ -701,7 +709,7 @@ def resolve_stack(self):
         if not callable(self.alias):
             return
         # check that we actual need the stack
-        sig = inspect.signature(self.alias)
+        sig = inspect.signature(getattr(self.alias, "func", self.alias))
         if len(sig.parameters) <= 5 and "stack" not in sig.parameters:
             return
         # compute the stack, and filter out these build methods

xonsh/tools.py

@@ -2858,3 +2858,8 @@ def unquote(s: str, chars="'\""):
     if len(s) >= 2 and s[0] == s[-1] and s[0] in chars:
         return s[1:-1]
     return s
+
+
+def endswith_newline(s: str):
+    """Add new line character to string if needed."""
+    return s if s.endswith("\n") else (s + "\n")