Source code for xonsh.contexts

"""Context management tools for xonsh."""

import sys
import textwrap
from collections.abc import Mapping

from xonsh.built_ins import XSH




class Block:
    """This is a context manager for obtaining a block of lines without actually
    executing the block. The lines are accessible as the 'lines' attribute.
    This must be used as a macro.
    """

    __xonsh_block__ = str

    def __init__(self):
        """
        Attributes
        ----------
        lines : list of str or None
            Block lines as if split by str.splitlines(), if available.
        glbs : Mapping or None
            Global execution context, ie globals().
        locs : Mapping or None
            Local execution context, ie locals().
        """
        self.lines = self.glbs = self.locs = None

    def __enter__(self):
        if not hasattr(self, "macro_block"):
            raise XSH.builtins.XonshError(
                self.__class__.__name__ + " must be entered as a macro!"
            )
        self.lines = self.macro_block.splitlines()
        self.glbs = self.macro_globals
        if self.macro_locals is not self.macro_globals:
            # leave locals as None when it is the same as globals
            self.locs = self.macro_locals
        return self

    def __exit__(self, exc_type, exc_value, traceback):
        pass





class Functor(Block):
    """This is a context manager that turns the block into a callable
    object, bound to the execution context it was created in.
    """

    def __init__(self, args=(), kwargs=None, rtn=""):
        """
        Parameters
        ----------
        args : Sequence of str, optional
            A tuple of argument names for the functor.
        kwargs : Mapping of str to values or list of item tuples, optional
            Keyword argument names and values, if available.
        rtn : str, optional
            Name of object to return, if available.

        Attributes
        ----------
        func : function
            The underlying function object. This defaults to none and is set
            after the the block is exited.
        """
        super().__init__()
        self.func = None
        self.args = args
        if kwargs is None:
            self.kwargs = []
        elif isinstance(kwargs, Mapping):
            self.kwargs = sorted(kwargs.items())
        else:
            self.kwargs = kwargs
        self.rtn = rtn

    def __enter__(self):
        super().__enter__()
        body = textwrap.indent(self.macro_block, "    ")
        uid = hash(body) + sys.maxsize  # should always be a positive int
        name = f"__xonsh_functor_{uid}__"
        # construct signature string
        sig = rtn = ""
        sig = ", ".join(self.args)
        kwstr = ", ".join([k + "=None" for k, _ in self.kwargs])
        if len(kwstr) > 0:
            sig = kwstr if len(sig) == 0 else sig + ", " + kwstr
        # construct return string
        rtn = str(self.rtn)
        if len(rtn) > 0:
            rtn = "    return " + rtn + "\n"
        # construct function string
        fstr = "def {name}({sig}):\n{body}\n{rtn}"
        fstr = fstr.format(name=name, sig=sig, body=body, rtn=rtn)
        glbs = self.glbs
        locs = self.locs
        execer = XSH.execer
        execer.exec(fstr, glbs=glbs, locs=locs)
        if locs is not None and name in locs:
            func = locs[name]
        elif name in glbs:
            func = glbs[name]
        else:
            raise ValueError("Functor block could not be found in context.")
        if len(self.kwargs) > 0:
            func.__defaults__ = tuple(v for _, v in self.kwargs)
        self.func = func
        return self

    def __exit__(self, exc_type, exc_value, traceback):
        pass

    def __call__(self, *args, **kwargs):
        """Dispatches to func."""
        if self.func is None:
            msg = "{} block with 'None' func not callable"
            raise AttributeError(msg.formst(self.__class__.__name__))
        return self.func(*args, **kwargs)