OILS / builtin / pure_ysh.py View on Github | oilshell.org

227 lines, 142 significant
1"""
2builtin/pure_ysh.py - YSH builtins that don't do I/O.
3"""
4from __future__ import print_function
5
6from _devbuild.gen.runtime_asdl import cmd_value
7from _devbuild.gen.syntax_asdl import command_t, loc, loc_t
8from _devbuild.gen.value_asdl import value, value_e, value_t
9from core import error
10from core import state
11from core import vm
12from frontend import flag_util
13from frontend import typed_args
14from mycpp import mylib
15from mycpp.mylib import tagswitch, NewDict
16
17from typing import TYPE_CHECKING, cast, Any, Dict, List
18
19if TYPE_CHECKING:
20 from display import ui
21 from osh.cmd_eval import CommandEvaluator
22
23
24class Shvar(vm._Builtin):
25
26 def __init__(self, mem, search_path, cmd_ev):
27 # type: (state.Mem, state.SearchPath, CommandEvaluator) -> None
28 self.mem = mem
29 self.search_path = search_path # to clear PATH
30 self.cmd_ev = cmd_ev # To run blocks
31
32 def Run(self, cmd_val):
33 # type: (cmd_value.Argv) -> int
34 _, arg_r = flag_util.ParseCmdVal('shvar',
35 cmd_val,
36 accept_typed_args=True)
37
38 # TODO: I think shvar LANG=C should just mutate
39 # But should there be a whitelist?
40 cmd_frag = typed_args.RequiredBlockAsFrag(cmd_val)
41
42 vars = NewDict() # type: Dict[str, value_t]
43 args, arg_locs = arg_r.Rest2()
44 if len(args) == 0:
45 raise error.Usage('Expected name=value', loc.Missing)
46
47 for i, arg in enumerate(args):
48 name, s = mylib.split_once(arg, '=')
49 if s is None:
50 raise error.Usage('Expected name=value', arg_locs[i])
51 v = value.Str(s) # type: value_t
52 vars[name] = v
53
54 # Important fix: shvar PATH='' { } must make all binaries invisible
55 if name == 'PATH':
56 self.search_path.ClearCache()
57
58 with state.ctx_Eval(self.mem, None, None, vars):
59 unused = self.cmd_ev.EvalCommandFrag(cmd_frag)
60
61 return 0
62
63
64class ctx_Context(object):
65 """For ctx push (context) { ... }"""
66
67 def __init__(self, mem, context):
68 # type: (state.Mem, Dict[str, value_t]) -> None
69 self.mem = mem
70 self.mem.PushContextStack(context)
71
72 def __enter__(self):
73 # type: () -> None
74 pass
75
76 def __exit__(self, type, value, traceback):
77 # type: (Any, Any, Any) -> None
78 self.mem.PopContextStack()
79
80
81class Ctx(vm._Builtin):
82
83 def __init__(self, mem, cmd_ev):
84 # type: (state.Mem, CommandEvaluator) -> None
85 self.mem = mem
86 self.cmd_ev = cmd_ev # To run blocks
87
88 def _GetContext(self):
89 # type: () -> Dict[str, value_t]
90 ctx = self.mem.GetContext()
91 if ctx is None:
92 raise error.Expr(
93 "Could not find a context. Did you forget to 'ctx push'?",
94 loc.Missing)
95 return ctx
96
97 def _Push(self, context, block):
98 # type: (Dict[str, value_t], command_t) -> int
99 with ctx_Context(self.mem, context):
100 return self.cmd_ev.EvalCommandFrag(block)
101
102 def _Set(self, updates):
103 # type: (Dict[str, value_t]) -> int
104 ctx = self._GetContext()
105 ctx.update(updates)
106 return 0
107
108 def _Emit(self, field, item, blame):
109 # type: (str, value_t, loc_t) -> int
110 ctx = self._GetContext()
111
112 if field not in ctx:
113 ctx[field] = value.List([])
114
115 UP_arr = ctx[field]
116 if UP_arr.tag() != value_e.List:
117 raise error.TypeErr(
118 UP_arr,
119 "Expected the context item '%s' to be a List" % (field), blame)
120
121 arr = cast(value.List, UP_arr)
122 arr.items.append(item)
123
124 return 0
125
126 def Run(self, cmd_val):
127 # type: (cmd_value.Argv) -> int
128 rd = typed_args.ReaderForProc(cmd_val)
129 _, arg_r = flag_util.ParseCmdVal('ctx',
130 cmd_val,
131 accept_typed_args=True)
132
133 verb, verb_loc = arg_r.ReadRequired2(
134 'Expected a verb (push, set, emit)')
135
136 if verb == "push":
137 context = rd.PosDict()
138 block = rd.RequiredBlockAsFrag()
139 rd.Done()
140 arg_r.AtEnd()
141
142 return self._Push(context, block)
143
144 elif verb == "set":
145 updates = rd.RestNamed()
146 rd.Done()
147 arg_r.AtEnd()
148
149 return self._Set(updates)
150
151 elif verb == "emit":
152 field, field_loc = arg_r.ReadRequired2(
153 "A target field is required")
154 item = rd.PosValue()
155 rd.Done()
156 arg_r.AtEnd()
157
158 return self._Emit(field, item, field_loc)
159
160 else:
161 raise error.Usage("Unknown verb '%s'" % verb, verb_loc)
162
163
164class PushRegisters(vm._Builtin):
165
166 def __init__(self, mem, cmd_ev):
167 # type: (state.Mem, CommandEvaluator) -> None
168 self.mem = mem
169 self.cmd_ev = cmd_ev # To run blocks
170
171 def Run(self, cmd_val):
172 # type: (cmd_value.Argv) -> int
173 _, arg_r = flag_util.ParseCmdVal('push-registers',
174 cmd_val,
175 accept_typed_args=True)
176
177 cmd_frag = typed_args.RequiredBlockAsFrag(cmd_val)
178
179 with state.ctx_Registers(self.mem):
180 unused = self.cmd_ev.EvalCommandFrag(cmd_frag)
181
182 # make it "SILENT" in terms of not mutating $?
183 # TODO: Revisit this. It might be better to provide the headless shell
184 # with a way to SET $? instead. Needs to be tested/prototyped.
185 return self.mem.last_status[-1]
186
187
188class Append(vm._Builtin):
189 """Push word args onto an List.
190
191 Not doing typed args since you can do
192
193 :: mylist->append(42)
194 """
195
196 def __init__(self, mem, errfmt):
197 # type: (state.Mem, ui.ErrorFormatter) -> None
198 self.mem = mem
199 self.errfmt = errfmt
200
201 def Run(self, cmd_val):
202 # type: (cmd_value.Argv) -> int
203
204 # This means we ignore -- , which is consistent
205 _, arg_r = flag_util.ParseCmdVal('append',
206 cmd_val,
207 accept_typed_args=True)
208
209 rd = typed_args.ReaderForProc(cmd_val)
210 val = rd.PosValue()
211 rd.Done()
212
213 UP_val = val
214 with tagswitch(val) as case:
215 if case(value_e.BashArray):
216 val = cast(value.BashArray, UP_val)
217 val.strs.extend(arg_r.Rest())
218 elif case(value_e.List):
219 val = cast(value.List, UP_val)
220 typed = [value.Str(s)
221 for s in arg_r.Rest()] # type: List[value_t]
222 val.items.extend(typed)
223 else:
224 raise error.TypeErr(val, 'expected List or BashArray',
225 loc.Missing)
226
227 return 0