def __init__(self, stmt, context):
self.stmt = stmt
self.context = context
self.stmt_table = {
ast.Expr: self.expr,
ast.Pass: self.parse_pass,
ast.AnnAssign: self.ann_assign,
ast.Assign: self.assign,
ast.If: self.parse_if,
ast.Call: self.call,
ast.Assert: self.parse_assert,
ast.For: self.parse_for,
ast.AugAssign: self.aug_assign,
ast.Break: self.parse_break,
ast.Return: self.parse_return,
}
stmt_type = self.stmt.__class__
if stmt_type in self.stmt_table:
self.lll_node = self.stmt_table[stmt_type]()
elif isinstance(stmt, ast.Name) and stmt.id == "throw":
self.lll_node = LLLnode.from_list(['assert', 0], typ=None, pos=getpos(stmt))
else:
raise StructureException("Unsupported statement type", stmt)
python类Return()的实例源码
def isDefault(a):
"""Our programs have a default setting of return 42, so we should detect that"""
if type(a) == ast.Module and len(a.body) == 1:
a = a.body[0]
else:
return False
if type(a) != ast.FunctionDef:
return False
if len(a.body) == 0:
return True
elif len(a.body) == 1:
if type(a.body[0]) == ast.Return:
if a.body[0].value == None or \
type(a.body[0].value) == ast.Num and a.body[0].value.n == 42:
return True
return False
def get_resource(name, connection=None):
"""
Return a tuple containing the Resource, Corpus, and Lexicon of the
corpus module specified by 'name'.
Arguments
---------
name : str
The name of the corpus module
connection : str or None
The name of the database connection. If None, the current connection
is used.
Returns
-------
res : tuple
A tuple consisting of the Resource class, Corpus class, and Lexicon
class defined in the corpus module
"""
if not connection:
connection = cfg.current_server
Resource, Corpus, Lexicon, _ = get_available_resources(connection)[name]
return Resource, Corpus, Lexicon
def get_unused_code(self, min_confidence=0, sort_by_size=False):
"""
Return ordered list of unused Item objects.
"""
if not 0 <= min_confidence <= 100:
raise ValueError('min_confidence must be between 0 and 100.')
def by_name(item):
return (item.filename.lower(), item.first_lineno)
def by_size(item):
return (item.size,) + by_name(item)
unused_code = (self.unused_attrs + self.unused_classes +
self.unused_funcs + self.unused_imports +
self.unused_props + self.unused_vars +
self.unreachable_code)
confidently_unused = [obj for obj in unused_code
if obj.confidence >= min_confidence]
return sorted(confidently_unused,
key=by_size if sort_by_size else by_name)
def _handle_ast_list(self, ast_list):
"""
Find unreachable nodes in the given sequence of ast nodes.
"""
for index, node in enumerate(ast_list):
if isinstance(node, (ast.Break, ast.Continue, ast.Raise,
ast.Return)):
try:
first_unreachable_node = ast_list[index + 1]
except IndexError:
continue
class_name = node.__class__.__name__.lower()
self._define(
self.unreachable_code,
class_name,
first_unreachable_node,
last_node=ast_list[-1],
message="unreachable code after '{class_name}'".format(
**locals()),
confidence=100)
return
def test_return_logger(self):
self.config.logger = io.StringIO()
def get_logs():
logger = self.config.logger
logger.seek(0)
return logger.readlines()
self.check_optimize("""
def func():
x = 1
return 1
return 2
""", """
def func():
x = 1
return 1
""")
self.assertEqual(get_logs(),
['<string>:4: fatoptimizer: Remove unreachable code: '
'Return(value=Constant(value=2))\n'])
def remove_dead_code(optimizer, node_list):
"""Remove dead code.
Modify node_list in-place.
Example: replace "return 1; return 2" with "return 1".
"""
truncate = None
stop = len(node_list) - 1
for index, node in enumerate(node_list):
if index == stop:
break
if not isinstance(node, (ast.Return, ast.Raise)):
continue
if not can_remove(node_list[index+1:]):
continue
truncate = index
break
# FIXME: use for/else: ?
if truncate is None:
return node_list
optimizer.log_node_removal("Remove unreachable code", node_list[truncate+1:])
return node_list[:truncate+1]
def build_positional_args(self, candidate, callsite):
"""Attempt to convert the positional and keyword args supplied at
the given callsite to the positional args expected by the candidate
funcdef.
Return a list of ast.Node instances, or raise ValueError if it
can't be done.
"""
if len(callsite.args) > len(candidate.args.args):
raise ValueError('too many positional arguments')
slots = {}
for idx, arg in enumerate(callsite.args):
slots[idx] = arg
for actual_kwarg in get_keywords(callsite):
idx = locate_kwarg(candidate, actual_kwarg.arg)
if idx in slots:
raise ValueError('positional slot %i already filled' % idx)
slots[idx] = actual_kwarg.value
actual_pos_args = []
for idx in range(len(candidate.args.args)):
if idx not in slots:
raise ValueError('argument %i not filled' % idx)
actual_pos_args.append(slots[idx])
return actual_pos_args
def callers(self, n=4, count=0, excludeCaller=True, files=False):
'''Return a list containing the callers of the function that called g.callerList.
If the excludeCaller keyword is True (the default), g.callers is not on the list.
If the files keyword argument is True, filenames are included in the list.
'''
# sys._getframe throws ValueError in both cpython and jython if there are less than i entries.
# The jython stack often has less than 8 entries,
# so we must be careful to call g._callerName with smaller values of i first.
result = []
i = 3 if excludeCaller else 2
while 1:
s = self._callerName(i, files=files)
# print(i,s)
if s:
result.append(s)
if not s or len(result) >= n: break
i += 1
result.reverse()
if count > 0: result = result[: count]
sep = '\n' if files else ','
return sep.join(result)
def all_matches(self, s):
'''
Return a list of match objects for all matches in s.
These are regex match objects or (start, end) for balanced searches.
'''
trace = False
if self.is_balanced():
aList, i = [], 0
while i < len(s):
progress = i
j = self.full_balanced_match(s, i)
if j is None:
i += 1
else:
aList.append((i,j),)
i = j
assert progress < i
return aList
else:
return list(self.regex.finditer(s))
def reduce_numbers(self, aList):
'''
Return aList with all number types in aList replaced by the most
general numeric type in aList.
'''
trace = False
found = None
numbers = ('number', 'complex', 'float', 'long', 'int')
for kind in numbers:
for z in aList:
if z == kind:
found = kind
break
if found:
break
if found:
assert found in numbers, found
aList = [z for z in aList if z not in numbers]
aList.append(found)
if trace: g.trace(aList)
return aList
def trace_stubs(self, stub, aList=None, header=None, level=-1):
'''Return a trace of the given stub and all its descendants.'''
indent = ' '*4*max(0,level)
if level == -1:
aList = ['===== %s...\n' % (header) if header else '']
for s in stub.out_list:
aList.append('%s%s' % (indent, s.rstrip()))
for child in stub.children:
self.trace_stubs(child, level=level+1, aList=aList)
if level == -1:
return '\n'.join(aList) + '\n'
# 2: ClassDef(identifier name, expr* bases,
# stmt* body, expr* decorator_list)
# 3: ClassDef(identifier name, expr* bases,
# keyword* keywords, expr? starargs, expr? kwargs
# stmt* body, expr* decorator_list)
#
# keyword arguments supplied to call (NULL identifier for **kwargs)
# keyword = (identifier? arg, expr value)
def generic_visit(self, node):
super(NodeTransformer, self).generic_visit(node)
if hasattr(node, 'body') and type(node.body) is list:
returns = [i for i, child in enumerate(node.body) if type(child) is ast.Return]
if len(returns) > 0:
for wait in self.get_waits():
node.body.insert(returns[0], wait)
inserts = []
for i, child in enumerate(node.body):
if type(child) is ast.Expr and self.is_concurrent_call(child.value):
self.encounter_call(child.value)
elif self.is_valid_assignment(child):
call = child.value
self.encounter_call(call)
name = child.targets[0].value
self.arguments.add(SchedulerRewriter.top_level_name(name))
index = child.targets[0].slice.value
call.func = ast.Attribute(call.func, 'assign', ast.Load())
call.args = [ast.Tuple([name, index], ast.Load())] + call.args
node.body[i] = ast.Expr(call)
elif self.references_arg(child):
inserts.insert(0, i)
for index in inserts:
for wait in self.get_waits():
node.body.insert(index, wait)
def get_coverable_nodes(cls):
return {
ast.Assert,
ast.Assign,
ast.AugAssign,
ast.Break,
ast.Continue,
ast.Delete,
ast.Expr,
ast.Global,
ast.Import,
ast.ImportFrom,
ast.Nonlocal,
ast.Pass,
ast.Raise,
ast.Return,
ast.FunctionDef,
ast.ClassDef,
ast.TryExcept,
ast.TryFinally,
ast.ExceptHandler,
ast.If,
ast.For,
ast.While,
}
def get_coverable_nodes(cls):
return {
ast.Assert,
ast.Assign,
ast.AugAssign,
ast.Break,
ast.Continue,
ast.Delete,
ast.Expr,
ast.Global,
ast.Import,
ast.ImportFrom,
ast.Nonlocal,
ast.Pass,
ast.Raise,
ast.Return,
ast.ClassDef,
ast.FunctionDef,
ast.Try,
ast.ExceptHandler,
ast.If,
ast.For,
ast.While,
}
def pythoneval(self, args, debug=False):
refs = [ast.Name("x{0}".format(i), ast.Load()) for i in xrange(len(args))]
if sys.version_info[0] <= 2:
params = ast.arguments([ast.Name("x{0}".format(i), ast.Param()) for i in xrange(len(args))], None, None, [])
fcn = ast.FunctionDef("tmp", params, [ast.Return(self.pythonast(refs))], [])
else:
params = ast.arguments([ast.arg("x{0}".format(i), None) for i in xrange(len(args))], None, [], [], None, [])
fcn = ast.FunctionDef("tmp", params, [ast.Return(self.pythonast(refs))], [], None)
moduleast = ast.Module([fcn])
fakeLineNumbers(moduleast)
if debug:
print(astToSource(fcn))
modulecomp = compile(moduleast, "Femtocode", "exec")
out = {"$importlib": importlib, "$math": math}
exec(modulecomp, out)
return out["tmp"](*args)
def translate_object(py_obj, body_only=False, enable_es6=False,
enable_stage3=False):
"""Translate the given Python 3 object (function, class, etc.) to ES6
Javascript.
If `body_only` is ``True``, the object itself is discarded and only its
body gets translated as it was a module body.
Return a ``(js_text, js_source_map)`` tuple.
"""
cwd = os.getcwd()
src_filename = os.path.abspath(inspect.getsourcefile(py_obj))
prefix = os.path.commonpath((cwd, src_filename))
if len(prefix) > 1:
src_filename = src_filename[len(prefix) + 1:]
src_lines, sline_offset = inspect.getsourcelines(py_obj)
# line offsets should be 0-based
sline_offset = sline_offset - 1
with open(src_filename) as f:
complete_src = f.read()
return translates(src_lines, True, src_filename, (sline_offset, 0),
body_only=body_only, complete_src=complete_src,
enable_es6=enable_es6, enable_stage3=enable_stage3)
def _build_node_cfg(node):
handlers = {
ast.If: _build_if_cfg,
ast.For: _build_loop_cfg,
ast.While: _build_loop_cfg,
ast.With: _build_with_cfg,
ast.Break: _build_break_cfg,
ast.Continue: _build_continue_cfg,
ast.Return: _build_return_cfg,
ast.Try: _build_try_cfg,
}
if type(node) in handlers:
handler = handlers[type(node)]
else:
handler = _build_statement_cfg
return handler(node)
def _inline(node, gen_sym, return_name, constants):
"""
Return a list of nodes, representing inlined function call.
"""
fn = constants[node.func.id]
fn_ast = Function.from_object(fn).tree
gen_sym, new_fn_ast = mangle(gen_sym, fn_ast)
parameter_assignments = _build_parameter_assignments(node, new_fn_ast)
body_nodes = new_fn_ast.body
gen_sym, inlined_body, new_bindings = _wrap_in_loop(gen_sym, body_nodes, return_name)
constants = dict(constants)
constants.update(new_bindings)
return parameter_assignments + inlined_body, gen_sym, constants
def filter_block(node_list):
"""
Remove no-op code (``pass``), or any code after
an unconditional jump (``return``, ``break``, ``continue``, ``raise``).
"""
if len(node_list) == 1:
return node_list
new_list = []
for node in node_list:
if type(node) == ast.Pass:
continue
new_list.append(node)
if type(node) in (ast.Return, ast.Break, ast.Continue, ast.Raise):
break
if len(new_list) == len(node_list):
return node_list
else:
return new_list
def init_inc_idx(cls, inc_idx):
if inc_idx == Ellipsis:
# if no index is provided, we will extract signature from the fn body
pass
elif isinstance(inc_idx, tuple):
inc_idx = cls._normalize_fn_idx(inc_idx)
(arg_types, return_type) = inc_idx
for arg_type in arg_types:
if not isinstance(arg_type, typy.Type):
raise typy.TypeFormationError(
"Argument type is not a type.")
if return_type != Ellipsis:
raise typy.TypeFormationError(
"Return type for an incomplete fn type must be elided.")
else:
raise typy.TypeFormationError(
"Incomplete fn type index is not an ellipsis or pair.")
return inc_idx
def check_for_b901(self, node):
xs = list(node.body)
has_yield = False
return_node = None
while xs:
x = xs.pop()
if isinstance(x, (ast.AsyncFunctionDef, ast.FunctionDef)):
continue
elif isinstance(x, (ast.Yield, ast.YieldFrom)):
has_yield = True
elif isinstance(x, ast.Return) and x.value is not None:
return_node = x
if has_yield and return_node is not None:
self.errors.append(
B901(return_node.lineno, return_node.col_offset)
)
break
xs.extend(ast.iter_child_nodes(x))
def visit_Lambda(self, t):
return Function('<lambda>', t.args, [ast.Return(t.body)])
def visit_ListComp(self, t):
result_append = ast.Attribute(ast.Name('.0', load), 'append', load)
body = ast.Expr(Call(result_append, [t.elt]))
for loop in reversed(t.generators):
for test in reversed(loop.ifs):
body = ast.If(test, [body], [])
body = ast.For(loop.target, loop.iter, [body], [])
fn = [body,
ast.Return(ast.Name('.0', load))]
args = ast.arguments([ast.arg('.0', None)], None, [], None, [], [])
return Call(Function('<listcomp>', args, fn),
[ast.List([], load)])
def visit_Lambda(self, t):
t = self.generic_visit(t)
result = Function('<lambda>', t.args, [ast.Return(t.body)])
return ast.copy_location(result, t)
def visit_ListComp(self, t):
t = self.generic_visit(t)
add_element = ast.Attribute(ast.Name('.elements', load), 'append', load)
body = ast.Expr(Call(add_element, [t.elt]))
for loop in reversed(t.generators):
for test in reversed(loop.ifs):
body = ast.If(test, [body], [])
body = ast.For(loop.target, loop.iter, [body], [])
fn = [body,
ast.Return(ast.Name('.elements', load))]
args = ast.arguments([ast.arg('.elements', None)], None, [], None, [], [])
result = Call(Function('<listcomp>', args, fn),
[ast.List([], load)])
return ast.copy_location(result, t)
def visit_Lambda(self, t):
t = self.generic_visit(t)
result = Function('<lambda>', t.args, [ast.Return(t.body)])
return ast.copy_location(result, t)
def visit_Lambda(self, t):
t = self.generic_visit(t)
result = Function('<lambda>', t.args, [ast.Return(t.body)])
return ast.copy_location(result, t)
def visit_ListComp(self, t):
t = self.generic_visit(t)
add_element = ast.Attribute(ast.Name('.elements', load), 'append', load)
body = ast.Expr(Call(add_element, [t.elt]))
for loop in reversed(t.generators):
for test in reversed(loop.ifs):
body = ast.If(test, [body], [])
body = ast.For(loop.target, loop.iter, [body], [])
fn = [body,
ast.Return(ast.Name('.elements', load))]
args = ast.arguments([ast.arg('.elements', None)], None, [], None, [], [])
result = Call(Function('<listcomp>', args, fn),
[ast.List([], load)])
return ast.copy_location(result, t)
def visit(self, node):
if self._return_level:
if self._node_level == self._return_level:
return None
elif self._node_level < self._return_level:
self._return_level = None
if isinstance(node, ast.Return):
self._return_level = self._node_level
self._node_level += 1
node = super().visit(node)
self._node_level -= 1
return node
