"""The xonsh built-ins.
Note that this module is named 'built_ins' so as not to be confused with the
special Python builtins module.
"""
from __future__ import annotations
import atexit
import builtins
import collections.abc as cabc
import contextlib
import inspect
import itertools
import os
import pathlib
import re
import signal
import sys
import types
import uuid
import warnings
from ast import AST
from collections.abc import Iterator
from operator import attrgetter as _attrgetter
from xonsh.lib.inspectors import Inspector
from xonsh.lib.lazyasd import lazyobject
from xonsh.platform import ON_POSIX, ON_WINDOWS
from xonsh.tools import (
XonshCalledProcessError,
XonshError,
expand_path,
globpath,
on_main_thread,
print_color,
)
INSPECTOR = Inspector()
warnings.filterwarnings("once", category=DeprecationWarning)
@lazyobject
def AT_EXIT_SIGNALS():
sigs = (
signal.SIGABRT,
signal.SIGFPE,
signal.SIGILL,
signal.SIGSEGV,
signal.SIGTERM,
)
if ON_POSIX:
sigs += (signal.SIGTSTP, signal.SIGQUIT, signal.SIGHUP)
return sigs
[docs]
def resetting_signal_handle(sig, f):
"""Sets a new signal handle that will automatically restore the old value
once the new handle is finished.
"""
# signal.signal() only works on the main thread; from a worker it raises
# ValueError. When xonsh is set up from a non-main thread (e.g. an embedded
# host calling xonsh.main.setup() off-main, like the 2020 virtualenv CI
# case in xonsh/xonsh#3689), skip registration rather than crash — the host
# owns process signals in that scenario. Same guard is used in procs/posix.py.
if not on_main_thread():
return
prev_signal_handler = signal.getsignal(sig)
def new_signal_handler(s=None, frame=None):
f(s, frame)
signal.signal(sig, prev_signal_handler)
if sig == signal.SIGHUP:
"""
SIGHUP means the controlling terminal has been lost. This should be
propagated to child processes so that they can decide what to do about it.
See also: https://www.gnu.org/software/bash/manual/bash.html#Signals
"""
import xonsh.procs.jobs as xj
xj.hup_all_jobs()
if sig != 0:
"""
There is no immediate exiting here.
The ``sys.exit()`` function raises a ``SystemExit`` exception.
This exception must be caught and processed in the upstream code.
"""
sys.exit(sig)
signal.signal(sig, new_signal_handler)
[docs]
def helper(x, name=""):
"""Prints help about, and then returns that variable."""
name = name or getattr(x, "__name__", "")
INSPECTOR.pinfo(x, oname=name, detail_level=0)
return x
[docs]
def superhelper(x, name=""):
"""Prints help about, and then returns that variable."""
name = name or getattr(x, "__name__", "")
INSPECTOR.pinfo(x, oname=name, detail_level=1)
return x
[docs]
def reglob(path, parts=None, i=None, include_dotfiles=False):
"""Regular expression-based globbing."""
if parts is None:
path = os.path.normpath(path)
drive, tail = os.path.splitdrive(path)
parts = tail.split(os.sep)
d = os.sep if os.path.isabs(path) else "."
d = os.path.join(drive, d)
return reglob(d, parts, i=0, include_dotfiles=include_dotfiles)
base = subdir = path
if i == 0:
if not os.path.isabs(base):
base = ""
elif len(parts) > 1:
i += 1
try:
regex = re.compile(parts[i])
except re.error as e:
original = "/".join(parts)
raise XonshError(
f"Regex glob error in segment {parts[i]!r} of pattern {original!r}: {e}. "
f"Regex globs are split by '/' and each segment is compiled separately, "
f"so groups and backreferences cannot span across '/'. "
f"See https://xon.sh/globbing.html"
) from e
files = os.listdir(subdir)
files.sort()
paths = []
i1 = i + 1
if i1 == len(parts):
for f in files:
if not include_dotfiles and f.startswith("."):
continue
p = os.path.join(base, f)
if regex.fullmatch(f) is not None:
paths.append(p)
else:
for f in files:
if not include_dotfiles and f.startswith("."):
continue
p = os.path.join(base, f)
if regex.fullmatch(f) is None or not os.path.isdir(p):
continue
paths += reglob(p, parts=parts, i=i1, include_dotfiles=include_dotfiles)
return paths
# mypy support
if sys.platform == "win32":
BasePath = pathlib.WindowsPath
else:
BasePath = pathlib.PosixPath
[docs]
class XonshPathLiteralChangeDirectoryContextManager:
"""Implements context manager to use in xonsh path literal."""
def __init__(self, path: XonshPathLiteral):
self.path = path
def __enter__(self):
self._xonsh_old_cwd = os.getcwd()
os.chdir(self.path)
return self.path
def __exit__(self, exc_type, exc, tb):
os.chdir(self._xonsh_old_cwd)
return False
[docs]
class XonshPathLiteral(BasePath): # type: ignore
"""Extension of ``pathlib.Path`` to support extended functionality."""
[docs]
def cd(self) -> XonshPathLiteralChangeDirectoryContextManager:
"""Returns context manager to change the directory
e.g. ``with p'/tmp'.cd(): $[ls]``
"""
return XonshPathLiteralChangeDirectoryContextManager(self)
[docs]
def mkdir(self, mode=0o777, parents=False, exist_ok=False):
"""Extension of ``pathlib.Path.mkdir`` that returns ``self`` instead of ``None``."""
super().mkdir(mode=mode, parents=parents, exist_ok=exist_ok)
return self
[docs]
def chmod(self, mode, *, follow_symlinks=True):
"""Extension of ``pathlib.Path.chmod`` that returns ``self`` instead of ``None``."""
super().chmod(mode, follow_symlinks=follow_symlinks)
return self
[docs]
def touch(self, mode=0o666, exist_ok=True):
"""Extension of ``pathlib.Path.touch`` that returns ``self`` instead of ``None``."""
super().touch(mode=mode, exist_ok=exist_ok)
return self
[docs]
class XonshList(list):
"""List subclass returned by glob operations with convenience methods.
All methods return XonshList, enabling chaining::
g`**/*.py`.files().sorted().paths()
"""
def _check_paths(self, method):
"""Raise if list contains tuples (e.g. from multi-group m``)."""
if self and isinstance(self[0], tuple):
raise TypeError(
f".{method}() requires string paths, got tuples. "
f"Use .select(n) to pick a tuple element first, e.g. .select(0).{method}()"
)
[docs]
def select(self, n):
"""Pick the n-th element from each tuple, skipping None values."""
return XonshList(
v
for x in self
for v in [x[n] if isinstance(x, tuple) else x]
if v is not None
)
[docs]
def unique(self):
"""Return a XonshList with duplicates removed, preserving order."""
return XonshList(dict.fromkeys(self))
[docs]
def paths(self):
"""Convert string elements to pathlib.Path objects."""
self._check_paths("paths")
return XonshList(pathlib.Path(p) for p in self)
[docs]
def sorted(self, key=None, reverse=False):
"""Return a new sorted XonshList."""
return XonshList(builtins.sorted(self, key=key, reverse=reverse))
[docs]
def filter(self, func):
"""Return a XonshList with only elements where func(elem) is truthy."""
return XonshList(x for x in self if func(x))
[docs]
def dirs(self):
"""Keep only paths that are existing directories."""
self._check_paths("dirs")
return XonshList(p for p in self if os.path.isdir(p))
[docs]
def files(self):
"""Keep only paths that are existing files."""
self._check_paths("files")
return XonshList(p for p in self if os.path.isfile(p))
[docs]
def exists(self):
"""Keep only paths that exist on disk."""
self._check_paths("exists")
return XonshList(p for p in self if os.path.exists(p))
@staticmethod
def _is_hidden(p):
"""Check if a path is hidden. Cross-platform: dotfiles on Unix,
FILE_ATTRIBUTE_HIDDEN on Windows."""
name = os.path.basename(p)
if name.startswith("."):
return True
if ON_WINDOWS:
try:
import stat
attrs = os.stat(p).st_file_attributes
return bool(attrs & stat.FILE_ATTRIBUTE_HIDDEN)
except (OSError, AttributeError):
pass
return False
[docs]
def hidden(self):
"""Keep only hidden files and directories.
On Unix: names starting with '.'. On Windows: also FILE_ATTRIBUTE_HIDDEN.
"""
self._check_paths("hidden")
return XonshList(p for p in self if self._is_hidden(p))
[docs]
def visible(self):
"""Keep only visible (non-hidden) files and directories.
On Unix: names not starting with '.'. On Windows: no FILE_ATTRIBUTE_HIDDEN.
"""
self._check_paths("visible")
return XonshList(p for p in self if not self._is_hidden(p))
[docs]
def path_literal(s):
s = expand_path(s)
return XonshPathLiteral(s)
[docs]
def regexsearch(s):
s = expand_path(s)
dotglob = XSH.env.get("DOTGLOB")
return XonshList(reglob(s, include_dotfiles=dotglob))
[docs]
def regexmatchsearch(s):
"""Regex glob that returns match groups instead of paths."""
s = expand_path(s)
dotglob = XSH.env.get("DOTGLOB")
regex = re.compile(s)
# Find the static prefix (path before any regex special chars)
_RE_SPECIAL = re.compile(r"[\\()\[\]{}.*+?|^$]")
m = _RE_SPECIAL.search(s)
if m:
prefix = s[: m.start()]
start = prefix.rsplit("/", 1)[0] or "."
else:
start = s if os.path.isdir(s) else os.path.dirname(s) or "."
results = []
for root, dirs, files in os.walk(start):
if not dotglob:
dirs[:] = [d for d in dirs if not d.startswith(".")]
for name in dirs + files:
if not dotglob and name.startswith("."):
continue
path = os.path.join(root, name)
match = regex.fullmatch(path)
if match:
groups = match.groups()
if not groups:
results.append(path)
elif len(groups) == 1:
results.append(groups[0])
else:
results.append(groups)
results.sort()
return XonshList(results)
[docs]
def globsearch(s):
glob_sorted = XSH.env.get("GLOB_SORTED")
dotglob = XSH.env.get("DOTGLOB")
return XonshList(
globpath(
s,
ignore_case=True,
return_empty=True,
sort_result=glob_sorted,
include_dotfiles=dotglob,
)
)
[docs]
def pathsearch(func, s, pymode=False, pathobj=False):
"""
Takes a string and returns a list of file paths that match (regex, glob,
or arbitrary search function). If pathobj=True, the return is a list of
pathlib.Path objects instead of strings.
"""
if not callable(func) or len(inspect.signature(func).parameters) != 1:
error = "%r is not a known path search function"
raise XonshError(error % func)
o = func(s)
if pathobj and pymode:
o = XonshList(map(pathlib.Path, o))
no_match = XonshList() if pymode else [s]
return o if len(o) != 0 else no_match
def _check_subproc_helper_raise(in_boolop):
"""Raise ``CalledProcessError`` if the most recently completed
pipeline (``XSH.lastcmd``) failed and the active settings say we
should raise.
Called by the ``subproc_*`` helpers right after they finish, so
that nested calls like ``echo @$(ls nono)`` raise on the inner
failure even though the *outer* helper would have succeeded with
an empty injection.
Skip conditions:
* ``in_boolop=True`` — the helper is a direct operand of a
``&&``/``||`` chain; let the chain wrapper see the result.
* ``$XONSH_SUBPROC_RAISE_ERROR`` is False.
* No completed pipeline (e.g. background job).
* The pipeline succeeded.
* ``spec.captured == 'object'`` — ``!(...)`` is the explicit
"user takes responsibility" form per spec.
* ``@error_ignore`` was applied (``spec.raise_subproc_error is False``).
"""
if in_boolop:
return
if not XSH.env.get("XONSH_SUBPROC_RAISE_ERROR"):
return
cp = XSH.lastcmd
if cp is None:
return
spec = getattr(cp, "spec", None)
if spec is None:
return
# Background job: pipeline is still running; reading .returncode would
# block on the blocking_property until the child exits.
if getattr(spec, "background", False):
return
if getattr(spec, "captured", None) == "object":
return
if getattr(spec, "raise_subproc_error", None) is False:
return
rtn = getattr(cp, "returncode", None)
if rtn is None or rtn == 0:
return
import subprocess
output = getattr(cp, "output", None)
raise subprocess.CalledProcessError(rtn, spec.args, output=output)
[docs]
def subproc_captured_stdout(*cmds, envs=None, in_boolop=False):
"""Runs a subprocess, capturing the output. Returns the stdout
that was produced as a str or list based on ``$XONSH_SUBPROC_OUTPUT_FORMAT``.
``in_boolop`` is set by the parser to True when this call is a direct
operand of a ``&&``/``||`` chain, so the runtime can adjust behavior
(e.g. suppress ``$XONSH_SUBPROC_CMD_RAISE_ERROR`` to let the chain
short-circuit on returncode).
"""
import xonsh.procs.specs
r = xonsh.procs.specs.run_subproc(
cmds, captured="stdout", envs=envs, in_boolop=in_boolop
)
_check_subproc_helper_raise(in_boolop)
return r
[docs]
def subproc_captured_inject(*cmds, envs=None, in_boolop=False):
"""Runs a subprocess, capturing the output. Returns a list of
whitespace-separated strings of the stdout that was produced.
The string is split using xonsh's lexer, rather than Python's str.split()
or shlex.split().
"""
import xonsh.procs.specs
o = xonsh.procs.specs.run_subproc(
cmds, captured="stdout", envs=envs, in_boolop=in_boolop
)
_check_subproc_helper_raise(in_boolop)
o = o if isinstance(o, list) else o.splitlines()
toks = []
for line in o:
line = line.rstrip(os.linesep)
toks.extend(XSH.execer.parser.lexer.split(line))
return toks
[docs]
def subproc_captured_object(*cmds, envs=None, in_boolop=False):
"""
Runs a subprocess, capturing the output. Returns an instance of
CommandPipeline representing the completed command.
"""
import xonsh.procs.specs
return xonsh.procs.specs.run_subproc(
cmds, captured="object", envs=envs, in_boolop=in_boolop
)
[docs]
def subproc_captured_hiddenobject(*cmds, envs=None, in_boolop=False):
"""Runs a subprocess, capturing the output. Returns an instance of
HiddenCommandPipeline representing the completed command.
"""
import xonsh.procs.specs
return xonsh.procs.specs.run_subproc(
cmds, captured="hiddenobject", envs=envs, in_boolop=in_boolop
)
[docs]
def subproc_uncaptured(*cmds, envs=None, in_boolop=False):
"""Runs a subprocess, without capturing the output. Returns the stdout
that was produced as a str.
"""
import xonsh.procs.specs
r = xonsh.procs.specs.run_subproc(
cmds, captured=False, envs=envs, in_boolop=in_boolop
)
_check_subproc_helper_raise(in_boolop)
return r
[docs]
def subproc_check_boolop(value):
"""Raise ``CalledProcessError`` if *value* is the (final) result of
a subproc statement that ended in failure.
Used as the AST wrapper around two kinds of nodes:
* The outermost ``BoolOp`` of a logical chain (``cmd1 && cmd2`` /
``cmd1 || cmd2``). Python ``and``/``or`` short-circuits to the
pipeline that determined the chain result, so the wrapper sees
the *final* pipeline.
* A standalone subproc-helper call at statement level — bare
commands, ``![...]``, ``$[...]``, ``$(...)``, ``@$(...)``. The
explicit-capture form ``!(...)`` is *intentionally* not wrapped:
per spec it is the user's full responsibility.
When ``$XONSH_SUBPROC_RAISE_ERROR`` is True (default), this enforces
"raise on the last executed pipeline's non-zero exit code" semantics:
* ``ls nono`` (standalone) — rc≠0: raises.
* ``$(ls nono)`` (standalone) — rc≠0: raises.
* ``$[ls nono]`` (standalone) — rc≠0: raises.
* ``echo 1 && echo 2`` — last is ``echo 2`` (rc=0): no raise.
* ``ls nono || echo 1`` — last is ``echo 1`` (rc=0): no raise.
* ``ls nono && echo 1`` — last is ``ls nono`` (rc≠0): raises.
Per-command overrides:
* ``@error_ignore`` (``spec.raise_subproc_error is False``) on the
final pipeline suppresses the raise — used e.g. for
``echo 1 && @error_ignore ls nono``.
* ``@error_raise`` raises directly at the pipeline level, so the
wrapper never sees a failed value.
Some subproc helpers (``$()``, ``$[]``, ``@$()``) return a *string*
/ *None* / *list* rather than a ``CommandPipeline``, so the wrapper
can't read ``returncode`` directly off ``value``. In that case it
falls back to ``XSH.lastcmd`` — the most recently completed pipeline,
which is the one this helper just ran.
"""
if not XSH.env.get("XONSH_SUBPROC_RAISE_ERROR"):
return value
# Try to get the pipeline from the value first (BoolOp result, ![...]).
spec = getattr(value, "spec", None)
# Background job: pipeline is still running; reading .returncode would
# block on the blocking_property until the child exits.
if spec is not None and getattr(spec, "background", False):
return value
rtn = getattr(value, "returncode", None)
cp_for_output = value
if spec is None or rtn is None:
# Value isn't a CommandPipeline (e.g. string from $(), None from
# $[], list from @$()). Fall back to the most recently
# completed pipeline so we can still inspect its returncode.
last = XSH.lastcmd
if last is None:
return value
spec = getattr(last, "spec", None)
if spec is not None and getattr(spec, "background", False):
return value
rtn = getattr(last, "returncode", None)
cp_for_output = last
if spec is None or rtn is None or rtn == 0:
return value
# ``!(...)`` is the "user takes full responsibility" form — never
# raise even if it ends up here via a chain like ``cmd && !(cmd)``.
if getattr(spec, "captured", None) == "object":
return value
# @error_ignore on the final pipeline of the chain opts out.
if getattr(spec, "raise_subproc_error", None) is False:
return value
import subprocess
output = getattr(cp_for_output, "output", None)
raise subprocess.CalledProcessError(rtn, spec.args, output=output)
_SPECIAL_BUILTINS = {"...": ...}
[docs]
def builtin_cmd(name):
"""Run a builtin name as a subprocess command if $XONSH_BUILTINS_TO_CMD is set
and the name is a known command or alias. Otherwise return the builtin value."""
env = XSH.env or {}
if env.get("XONSH_BUILTINS_TO_CMD"):
has_cmd = name in (XSH.aliases or {})
if not has_cmd and XSH.commands_cache:
has_cmd = XSH.commands_cache.locate_binary(name) is not None
if has_cmd:
return subproc_captured_hiddenobject([name])
if name in _SPECIAL_BUILTINS:
return _SPECIAL_BUILTINS[name]
import builtins as _builtins
return getattr(_builtins, name, None)
[docs]
def ensure_list_of_strs(x):
"""Ensures that x is a list of strings."""
if isinstance(x, str):
rtn = [x]
elif isinstance(x, cabc.Sequence):
rtn = [i if isinstance(i, str) else str(i) for i in x]
else:
rtn = [str(x)]
return rtn
[docs]
def ensure_str_or_callable(x):
"""Ensures that x is single string or function."""
if isinstance(x, str) or callable(x):
return x
if isinstance(x, bytes):
# ``os.fsdecode`` decodes using "surrogateescape" on linux and "strict" on windows.
# This is used to decode bytes for interfacing with the os, notably for command line arguments.
# See https://www.python.org/dev/peps/pep-0383/#specification
return os.fsdecode(x)
return str(x)
[docs]
def list_of_strs_or_callables(x):
"""
Ensures that x is a list of strings or functions.
This is called when using the ``@()`` operator to expand it's content.
"""
if isinstance(x, str | bytes) or callable(x):
rtn = [ensure_str_or_callable(x)]
elif isinstance(x, cabc.Iterable):
rtn = list(map(ensure_str_or_callable, x))
else:
rtn = [ensure_str_or_callable(x)]
return rtn
[docs]
def list_of_list_of_strs_outer_product(x):
"""Takes an outer product of a list of strings"""
lolos = map(ensure_list_of_strs, x)
rtn = []
for los in itertools.product(*lolos):
s = "".join(los)
if "*" in s:
rtn.extend(XSH.glob(s))
else:
rtn.append(XSH.expand_path(s))
return rtn
[docs]
def eval_fstring_field(field):
"""Evaluates the argument in Xonsh context."""
res = XSH.execer.eval(
field[0].strip(), glbs=globals(), locs=XSH.ctx, filename=field[1]
)
return res
@lazyobject
def MACRO_FLAG_KINDS():
return {
"s": str,
"str": str,
"string": str,
"a": AST,
"ast": AST,
"c": types.CodeType,
"code": types.CodeType,
"compile": types.CodeType,
"v": eval,
"eval": eval,
"x": exec,
"exec": exec,
"t": type,
"type": type,
}
def _convert_kind_flag(x):
"""Puts a kind flag (string) a canonical form."""
x = x.lower()
kind = MACRO_FLAG_KINDS.get(x, None)
if kind is None:
raise TypeError(f"{x!r} not a recognized macro type.")
return kind
[docs]
def convert_macro_arg(raw_arg, kind, glbs, locs, *, name="<arg>", macroname="<macro>"):
"""Converts a string macro argument based on the requested kind.
Parameters
----------
raw_arg : str
The str representation of the macro argument.
kind : object
A flag or type representing how to convert the argument.
glbs : Mapping
The globals from the call site.
locs : Mapping or None
The locals from the call site.
name : str, optional
The macro argument name.
macroname : str, optional
The name of the macro itself.
Returns
-------
The converted argument.
"""
# munge kind and mode to start
mode = None
if isinstance(kind, cabc.Sequence) and not isinstance(kind, str):
# have (kind, mode) tuple
kind, mode = kind
if isinstance(kind, str):
kind = _convert_kind_flag(kind)
if kind is str or kind is None:
return raw_arg # short circuit since there is nothing else to do
# select from kind and convert
execer = XSH.execer
filename = macroname + "(" + name + ")"
if kind is AST:
ctx = set(dir(builtins)) | set(glbs.keys())
if locs is not None:
ctx |= set(locs.keys())
mode = mode or "eval"
if mode != "eval" and not raw_arg.endswith("\n"):
raw_arg += "\n"
arg = execer.parse(raw_arg, ctx, mode=mode, filename=filename)
elif kind is types.CodeType or kind is compile: # NOQA
mode = mode or "eval"
arg = execer.compile(
raw_arg, mode=mode, glbs=glbs, locs=locs, filename=filename
)
elif kind is eval:
arg = execer.eval(raw_arg, glbs=glbs, locs=locs, filename=filename)
elif kind is exec:
mode = mode or "exec"
if not raw_arg.endswith("\n"):
raw_arg += "\n"
arg = execer.exec(raw_arg, mode=mode, glbs=glbs, locs=locs, filename=filename)
elif kind is type:
arg = type(execer.eval(raw_arg, glbs=glbs, locs=locs, filename=filename))
else:
msg = "kind={0!r} and mode={1!r} was not recognized for macro argument {2!r}"
raise TypeError(msg.format(kind, mode, name))
return arg
[docs]
@contextlib.contextmanager
def in_macro_call(f, glbs, locs):
"""Attaches macro globals and locals temporarily to function as a
context manager.
Parameters
----------
f : callable object
The function that is called as ``f(*args)``.
glbs : Mapping
The globals from the call site.
locs : Mapping or None
The locals from the call site.
"""
prev_glbs = getattr(f, "macro_globals", None)
prev_locs = getattr(f, "macro_locals", None)
f.macro_globals = glbs
f.macro_locals = locs
yield
if prev_glbs is None:
del f.macro_globals
else:
f.macro_globals = prev_glbs
if prev_locs is None:
del f.macro_locals
else:
f.macro_locals = prev_locs
[docs]
def call_macro(f, raw_args, glbs, locs):
"""Calls a function as a macro, returning its result.
Parameters
----------
f : callable object
The function that is called as ``f(*args)``.
raw_args : tuple of str
The str representation of arguments of that were passed into the
macro. These strings will be parsed, compiled, evaled, or left as
a string depending on the annotations of f.
glbs : Mapping
The globals from the call site.
locs : Mapping or None
The locals from the call site.
"""
sig = inspect.signature(f)
empty = inspect.Parameter.empty
macroname = f.__name__
i = 0
args = []
for (key, param), raw_arg in zip(sig.parameters.items(), raw_args, strict=False):
i += 1
if raw_arg == "*":
break
kind = param.annotation
if kind is empty or kind is None:
kind = str
arg = convert_macro_arg(
raw_arg, kind, glbs, locs, name=key, macroname=macroname
)
args.append(arg)
reg_args, kwargs = _eval_regular_args(raw_args[i:], glbs, locs)
args += reg_args
with in_macro_call(f, glbs, locs):
rtn = f(*args, **kwargs)
return rtn
@lazyobject
def KWARG_RE():
return re.compile(r"([A-Za-z_]\w*=|\*\*)")
def _starts_as_arg(s):
"""Tests if a string starts as a non-kwarg string would."""
return KWARG_RE.match(s) is None
def _eval_regular_args(raw_args, glbs, locs):
if not raw_args:
return [], {}
arglist = list(itertools.takewhile(_starts_as_arg, raw_args))
kwarglist = raw_args[len(arglist) :]
execer = XSH.execer
if not arglist:
args = arglist
kwargstr = "dict({})".format(", ".join(kwarglist))
kwargs = execer.eval(kwargstr, glbs=glbs, locs=locs)
elif not kwarglist:
argstr = "({},)".format(", ".join(arglist))
args = execer.eval(argstr, glbs=glbs, locs=locs)
kwargs = {}
else:
argstr = "({},)".format(", ".join(arglist))
kwargstr = "dict({})".format(", ".join(kwarglist))
both = f"({argstr}, {kwargstr})"
args, kwargs = execer.eval(both, glbs=glbs, locs=locs)
return args, kwargs
[docs]
def enter_macro(obj, raw_block, glbs, locs):
"""Prepares to enter a context manager macro by attaching the contents
of the macro block, globals, and locals to the object. These modifications
are made in-place and the original object is returned.
Parameters
----------
obj : context manager
The object that is about to be entered via a with-statement.
raw_block : str
The str of the block that is the context body.
This string will be parsed, compiled, evaled, or left as
a string depending on the return annotation of obj.__enter__.
glbs : Mapping
The globals from the context site.
locs : Mapping or None
The locals from the context site.
Returns
-------
obj : context manager
The same context manager but with the new macro information applied.
"""
# recurse down sequences
if isinstance(obj, cabc.Sequence):
for x in obj:
enter_macro(x, raw_block, glbs, locs)
return obj
# convert block as needed
kind = getattr(obj, "__xonsh_block__", str)
macroname = getattr(obj, "__name__", "<context>")
block = convert_macro_arg(
raw_block, kind, glbs, locs, name="<with!>", macroname=macroname
)
# attach attrs
obj.macro_globals = glbs
obj.macro_locals = locs
obj.macro_block = block
return obj
[docs]
@contextlib.contextmanager
def xonsh_builtins(execer=None):
"""A context manager for using the xonsh builtins only in a limited
scope. Likely useful in testing.
"""
XSH.load(execer=execer)
yield
XSH.unload()
[docs]
class InlineImporter:
"""Inline importer allows to import and use module attribute or function in one line."""
def __getattr__(self, name):
if name.startswith("__"):
return getattr(super(), name)
return __import__(name)
[docs]
class Cmd:
"""A command group."""
def __init__(
self,
xsh: XonshSession,
*args: str,
bg=False,
redirects: dict[str, str] | None = None,
):
self.xsh = xsh
self.args: list[list[str | tuple[str, str]] | str] = []
self._add_proc(*args, redirects=redirects or {})
if bg:
self.args.append("&")
def _expand(self, *args: str | list[str]) -> Iterator[str]:
for arg in args:
if isinstance(arg, str):
yield expand_path(arg)
else:
yield from (expand_path(str(a)) for a in arg)
def _add_proc(self, *args: str, redirects: dict[str, str] | None = None) -> None:
"""a single Popen process args"""
cmds: list[str | tuple[str, str]] = list(self._expand(*args))
if redirects:
for k, v in redirects.items():
cmds.append((k, expand_path(v)))
self.args.append(cmds)
[docs]
def out(self):
"""dispatch $()"""
return self.xsh.subproc_captured_stdout(*self.args)
[docs]
def run(self):
"""dispatch $[]"""
return self.xsh.subproc_uncaptured(*self.args)
[docs]
def hide(self):
"""dispatch ![]"""
return self.xsh.subproc_captured_hiddenobject(*self.args)
[docs]
def obj(self):
"""dispatch !()"""
return self.xsh.subproc_captured_object(*self.args)
[docs]
def pipe(self, *args):
"""combine $() | $[]"""
self.args.append("|")
self._add_proc(*args)
return self
[docs]
class XonshSessionInterface:
"""Xonsh Session Interface
Attributes
----------
env : xonsh.environ.Env
A xonsh environment e.g. `@.env.get('HOME', '/tmp')`.
history : xonsh.history.History
Xonsh history backend e.g. `@.history[-1].cmd`.
See also `history --help` to manage history from
command line.
imp : xonsh.built_ins.InlineImporter
The inline importer provides instant access to library
functions and attributes e.g. `@.imp.time.time()`.
lastcmd : xonsh.procs.pipelines.CommandPipeline
Last executed subprocess-mode command pipeline
e.g. `@.lastcmd.rtn` returns exit code.
"""
env = None # type: ignore
history = None # type: ignore
imp: InlineImporter = InlineImporter()
lastcmd = None # type: ignore
[docs]
class XonshSession:
"""All components defining a xonsh session.
Warning! If you use this object for any reason and access ``__xonsh__``
or ``xonsh.built_ins.XSH`` attributes or functions, you do so at your
own risk, as the internal contents and behavior of this object may
change with any release. For repeatable use cases, find a way
to improve ``XonshSessionInterface`` or ``xonsh.api``.
"""
def __init__(self):
"""
Attributes
----------
exit: int or None
Session attribute. In case of integer value it signals xonsh to exit
with returning this value as exit code.
"""
self.interface = XonshSessionInterface()
self.execer = None
self.ctx = {}
self.builtins_loaded = False
self.history = None
self.shell = None
self.env = None
self.imp = InlineImporter()
self.rc_files = None
# AST-invoked functions
self.help = helper
self.superhelp = superhelper
self.pathsearch = pathsearch
self.globsearch = globsearch
self.regexsearch = regexsearch
self.regexmatchsearch = regexmatchsearch
self.glob = lambda *a, **kw: XonshList(globpath(*a, **kw))
self.expand_path = expand_path
self.subproc_captured_stdout = subproc_captured_stdout
self.subproc_captured_inject = subproc_captured_inject
self.subproc_captured_object = subproc_captured_object
self.subproc_captured_hiddenobject = subproc_captured_hiddenobject
self.subproc_uncaptured = subproc_uncaptured
self.subproc_check_boolop = subproc_check_boolop
self.call_macro = call_macro
self.enter_macro = enter_macro
self.path_literal = path_literal
self.builtin_cmd = builtin_cmd
self.list_of_strs_or_callables = list_of_strs_or_callables
self.list_of_list_of_strs_outer_product = list_of_list_of_strs_outer_product
self.eval_fstring_field = eval_fstring_field
# Session attributes
self.exit = None
self.stdout_uncaptured = None
self.stderr_uncaptured = None
self._py_exit = None
self._py_quit = None
self.commands_cache = None
self.modules_cache = None
self.all_jobs = None
self._completers = None
self.builtins = None
self._initial_builtin_names = None
self.lastcmd = None
self._last = None
self.sessionid = str(uuid.uuid4())
@property
def last(self):
warnings.warn(
"The `last` attribute is deprecated and will be removed. Use `lastcmd`.",
DeprecationWarning,
stacklevel=2,
)
return self._last
@last.setter
def last(self, value):
self._last = value
[docs]
def cmd(self, *args: str, **kwargs):
return Cmd(self, *args, **kwargs)
@property
def aliases(self):
if self.commands_cache is None:
return
return self.commands_cache.aliases
@property
def completers(self):
"""Returns a list of all available completers. Init when first accessing the attribute"""
if self._completers is None:
from xonsh.completers.init import default_completers
self._completers = default_completers(self.commands_cache)
return self._completers
def _disable_python_exit(self):
# Disable Python interactive quit/exit
if hasattr(builtins, "exit"):
self._py_exit = builtins.exit
del builtins.exit
if hasattr(builtins, "quit"):
self._py_quit = builtins.quit
del builtins.quit
def _restore_python_exit(self):
if self._py_exit is not None:
builtins.exit = self._py_exit
if self._py_quit is not None:
builtins.quit = self._py_quit
[docs]
def load(
self,
execer=None,
ctx=None,
inherit_env=True,
save_origin_env=False,
**kwargs,
):
"""Loads the session with default values.
Parameters
----------
execer : Execer, optional
Xonsh execution object, may be None to start
ctx : Mapping, optional
Context to start xonsh session with.
inherit_env : bool
If ``True``: inherit environment variables from ``os.environ``.
If ``False``: use default values for environment variables and
set ``$XONSH_ENV_INHERITED = False``.
"""
from xonsh.commands_cache import CommandsCache
from xonsh.environ import Env, default_env, save_origin_env_to_file
if not hasattr(builtins, "__xonsh__"):
builtins.__xonsh__ = self
if ctx is not None:
self.ctx = ctx
if "env" in kwargs:
self.env = kwargs.pop("env")
elif inherit_env:
self.env = Env(default_env())
else:
no_env = {"XONSH_ENV_INHERITED": False}
# TERM is essential for proper terminal operation
no_env["TERM"] = os.environ.get("TERM", "xterm")
self.env = Env(no_env)
self.interface.env = self.env
if save_origin_env:
save_origin_env_to_file(self.env, self.sessionid)
self.exit = None
self.stdout_uncaptured = None
self.stderr_uncaptured = None
self._disable_python_exit()
self.execer = execer
self.modules_cache = {}
self.all_jobs = {}
self.builtins = get_default_builtins(execer)
self._initial_builtin_names = frozenset(vars(self.builtins))
aliases_given = kwargs.pop("aliases", None)
for attr, value in kwargs.items():
if hasattr(self, attr):
setattr(self, attr, value)
self.commands_cache = (
kwargs.pop("commands_cache")
if "commands_cache" in kwargs
else CommandsCache(self.env, aliases_given)
)
self.link_builtins()
self.builtins_loaded = True
def flush_on_exit(s=None, f=None):
if self.history is not None:
self.history.flush(at_exit=True)
self._flush_on_exit = flush_on_exit
atexit.register(flush_on_exit)
# Add one-shot handler for exit
for sig in AT_EXIT_SIGNALS:
resetting_signal_handle(sig, flush_on_exit)
[docs]
def link_builtins(self):
# public built-ins
for refname in self._initial_builtin_names:
objname = f"__xonsh__.builtins.{refname}"
proxy = DynamicAccessProxy(refname, objname)
setattr(builtins, refname, proxy)
# sneak the path search functions into the aliases
# Need this inline/lazy import here since we use locate_binary that
# relies on __xonsh__.env in default aliases
builtins.default_aliases = builtins.aliases = self.aliases
[docs]
def unlink_builtins(self):
for name in self._initial_builtin_names:
if hasattr(builtins, name):
delattr(builtins, name)
[docs]
def unload(self):
if not hasattr(builtins, "__xonsh__"):
self.builtins_loaded = False
return
if hasattr(self.env, "undo_replace_env"):
self.env.undo_replace_env()
self._restore_python_exit()
if not self.builtins_loaded:
return
if self.history is not None:
self.history.flush(at_exit=True)
if hasattr(self, "_flush_on_exit"):
atexit.unregister(self._flush_on_exit)
self.unlink_builtins()
delattr(builtins, "__xonsh__")
self.builtins_loaded = False
self._completers = None
[docs]
def get_default_builtins(execer=None):
from xonsh.events import events
return types.SimpleNamespace(
XonshError=XonshError,
XonshCalledProcessError=XonshCalledProcessError,
evalx=None if execer is None else execer.eval,
execx=None if execer is None else execer.exec,
compilex=None if execer is None else execer.compile,
events=events,
print_color=print_color,
printx=print_color,
)
[docs]
class DynamicAccessProxy:
"""Proxies access dynamically."""
def __init__(self, refname, objname):
"""
Parameters
----------
refname : str
'.'-separated string that represents the new, reference name that
the user will access.
objname : str
'.'-separated string that represents the name where the target
object actually lives that refname points to.
"""
super().__setattr__("refname", refname)
super().__setattr__("objname", objname)
def _obj(self):
"""Dynamically grabs object.
This is a method instead of a ``@property`` as a workaround for
a Nuitka bugs.
- https://github.com/Nuitka/Nuitka/issues/3859
- https://github.com/Nuitka/Nuitka/issues/3860
When the bug is fixed upstream, this can be reverted to
``@property def obj(self)``.
Uses ``object.__getattribute__`` to read *objname* so that the
lookup never falls through to ``__getattr__`` — which would
create an infinite loop under Nuitka-compiled builds.
Uses :func:`operator.attrgetter` instead of an explicit ``getattr``
loop: the loop form compiled under Nuitka 4.0 produces wrong
results for ``getattr(builtins, "__xonsh__")`` (it doesn't advance
``obj`` on the first iteration), yielding ``AttributeError:
module 'builtins' has no attribute 'builtins'``. ``attrgetter``
delegates the whole walk to a C builtin and sidesteps the bug.
"""
return _attrgetter(object.__getattribute__(self, "objname"))(builtins)
def __getattr__(self, name):
return getattr(self._obj(), name)
def __setattr__(self, name, value):
return setattr(self._obj(), name, value)
def __delattr__(self, name):
return delattr(self._obj(), name)
def __getitem__(self, item):
return self._obj().__getitem__(item)
def __setitem__(self, item, value):
return self._obj().__setitem__(item, value)
def __delitem__(self, item):
del self._obj()[item]
def __call__(self, *args, **kwargs):
return self._obj().__call__(*args, **kwargs)
def __dir__(self):
return self._obj().__dir__()
def __repr__(self):
return repr(self._obj())
def __str__(self):
return str(self._obj())
# singleton
XSH = XonshSession()