def filterstr_to_filterfunc(filter_str: str, logged_in: bool) -> Callable[['Post'], bool]:
"""Takes an --only-if=... filter specification and makes a filter_func Callable out of it."""
# The filter_str is parsed, then all names occurring in its AST are replaced by loads to post.<name>. A
# function Post->bool is returned which evaluates the filter with the post as 'post' in its namespace.
class TransformFilterAst(ast.NodeTransformer):
def visit_Name(self, node: ast.Name):
# pylint:disable=invalid-name,no-self-use
if not isinstance(node.ctx, ast.Load):
raise InvalidArgumentException("Invalid filter: Modifying variables ({}) not allowed.".format(node.id))
if not hasattr(Post, node.id):
raise InvalidArgumentException("Invalid filter: Name {} is not defined.".format(node.id))
if node.id in Post.LOGIN_REQUIRING_PROPERTIES and not logged_in:
raise InvalidArgumentException("Invalid filter: Name {} requires being logged in.".format(node.id))
new_node = ast.Attribute(ast.copy_location(ast.Name('post', ast.Load()), node), node.id,
ast.copy_location(ast.Load(), node))
return ast.copy_location(new_node, node)
input_filename = '<--only-if parameter>'
compiled_filter = compile(TransformFilterAst().visit(ast.parse(filter_str, filename=input_filename, mode='eval')),
filename=input_filename, mode='eval')
def filterfunc(post: 'Post') -> bool:
# pylint:disable=eval-used
return bool(eval(compiled_filter, {'post': post}))
return filterfunc
python类Load()的实例源码
def pop_format_context(self, expl_expr):
"""Format the %-formatted string with current format context.
The expl_expr should be an ast.Str instance constructed from
the %-placeholders created by .explanation_param(). This will
add the required code to format said string to .on_failure and
return the ast.Name instance of the formatted string.
"""
current = self.stack.pop()
if self.stack:
self.explanation_specifiers = self.stack[-1]
keys = [ast.Str(key) for key in current.keys()]
format_dict = ast.Dict(keys, list(current.values()))
form = ast.BinOp(expl_expr, ast.Mod(), format_dict)
name = "@py_format" + str(next(self.variable_counter))
self.on_failure.append(ast.Assign([ast.Name(name, ast.Store())], form))
return ast.Name(name, ast.Load())
def checkGlobalIds(a, l):
if not isinstance(a, ast.AST):
return
elif type(a) in [ ast.Load, ast.Store, ast.Del, ast.AugLoad, ast.AugStore, ast.Param ]:
return
if not hasattr(a, "global_id"):
addedNodes = ["propagatedVariable", "orderedBinOp",
"augAssignVal", "augAssignBinOp",
"combinedConditional", "combinedConditionalOp",
"multiCompPart", "multiCompOp",
"second_global_id", "moved_line",
# above this line has individualize functions. below does not.
"addedNot", "addedNotOp", "addedOther", "addedOtherOp",
"collapsedExpr", "removedLines",
"helperVar", "helperReturn",
"typeCastFunction", ]
for t in addedNodes:
if hasattr(a, t):
break
else: # only enter the else if none of the provided types are an attribute of a
log("canonicalize\tcheckGlobalIds\tNo global id: " + str(l) + "," + str(a.__dict__) + "," + printFunction(a, 0), "bug")
for f in ast.iter_child_nodes(a):
checkGlobalIds(f, l + [type(a)])
def stateful_get_all_emojis(self, expr):
if not isinstance(expr.value, ast.Await):
return expr
if not isinstance(expr.value.value, ast.Call):
return expr
call = expr.value.value
if isinstance(call.func, ast.Attribute):
if call.func.attr == 'get_all_emojis':
if self.interactive and not prompt_change(
'A possible change was found to make get_all_emojis stateful.'
):
return expr
new_expr = ast.Expr()
new_expr.value = ast.Attribute()
new_expr.value.value = call.func.value
new_expr.value.attr = 'emojis'
new_expr.value.ctx = ast.Load()
new_expr = ast.copy_location(new_expr, expr)
stats_counter['expr_changes'] += 1
return new_expr
return expr
def test_dump(self):
node = ast.parse('spam(eggs, "and cheese")')
self.assertEqual(ast.dump(node),
"Module(body=[Expr(value=Call(func=Name(id='spam', ctx=Load()), "
"args=[Name(id='eggs', ctx=Load()), Str(s='and cheese')], "
"keywords=[], starargs=None, kwargs=None))])"
)
self.assertEqual(ast.dump(node, annotate_fields=False),
"Module([Expr(Call(Name('spam', Load()), [Name('eggs', Load()), "
"Str('and cheese')], [], None, None))])"
)
self.assertEqual(ast.dump(node, include_attributes=True),
"Module(body=[Expr(value=Call(func=Name(id='spam', ctx=Load(), "
"lineno=1, col_offset=0), args=[Name(id='eggs', ctx=Load(), "
"lineno=1, col_offset=5), Str(s='and cheese', lineno=1, "
"col_offset=11)], keywords=[], starargs=None, kwargs=None, "
"lineno=1, col_offset=0), lineno=1, col_offset=0)])"
)
def test_dump(self):
node = ast.parse('spam(eggs, "and cheese")')
self.assertEqual(ast.dump(node),
"Module(body=[Expr(value=Call(func=Name(id='spam', ctx=Load()), "
"args=[Name(id='eggs', ctx=Load()), Str(s='and cheese')], "
"keywords=[], starargs=None, kwargs=None))])"
)
self.assertEqual(ast.dump(node, annotate_fields=False),
"Module([Expr(Call(Name('spam', Load()), [Name('eggs', Load()), "
"Str('and cheese')], [], None, None))])"
)
self.assertEqual(ast.dump(node, include_attributes=True),
"Module(body=[Expr(value=Call(func=Name(id='spam', ctx=Load(), "
"lineno=1, col_offset=0), args=[Name(id='eggs', ctx=Load(), "
"lineno=1, col_offset=5), Str(s='and cheese', lineno=1, "
"col_offset=11)], keywords=[], starargs=None, kwargs=None, "
"lineno=1, col_offset=0), lineno=1, col_offset=0)])"
)
def test_dump(self):
node = ast.parse('spam(eggs, "and cheese")')
self.assertEqual(ast.dump(node),
"Module(body=[Expr(value=Call(func=Name(id='spam', ctx=Load()), "
"args=[Name(id='eggs', ctx=Load()), Str(s='and cheese')], "
"keywords=[], starargs=None, kwargs=None))])"
)
self.assertEqual(ast.dump(node, annotate_fields=False),
"Module([Expr(Call(Name('spam', Load()), [Name('eggs', Load()), "
"Str('and cheese')], [], None, None))])"
)
self.assertEqual(ast.dump(node, include_attributes=True),
"Module(body=[Expr(value=Call(func=Name(id='spam', ctx=Load(), "
"lineno=1, col_offset=0), args=[Name(id='eggs', ctx=Load(), "
"lineno=1, col_offset=5), Str(s='and cheese', lineno=1, "
"col_offset=11)], keywords=[], starargs=None, kwargs=None, "
"lineno=1, col_offset=0), lineno=1, col_offset=0)])"
)
def pop_format_context(self, expl_expr):
"""Format the %-formatted string with current format context.
The expl_expr should be an ast.Str instance constructed from
the %-placeholders created by .explanation_param(). This will
add the required code to format said string to .on_failure and
return the ast.Name instance of the formatted string.
"""
current = self.stack.pop()
if self.stack:
self.explanation_specifiers = self.stack[-1]
keys = [ast.Str(key) for key in current.keys()]
format_dict = ast.Dict(keys, list(current.values()))
form = ast.BinOp(expl_expr, ast.Mod(), format_dict)
name = "@py_format" + str(next(self.variable_counter))
self.on_failure.append(ast.Assign([ast.Name(name, ast.Store())], form))
return ast.Name(name, ast.Load())
def test_dump(self):
node = ast.parse('spam(eggs, "and cheese")')
self.assertEqual(ast.dump(node),
"Module(body=[Expr(value=Call(func=Name(id='spam', ctx=Load()), "
"args=[Name(id='eggs', ctx=Load()), Str(s='and cheese')], "
"keywords=[], starargs=None, kwargs=None))])"
)
self.assertEqual(ast.dump(node, annotate_fields=False),
"Module([Expr(Call(Name('spam', Load()), [Name('eggs', Load()), "
"Str('and cheese')], [], None, None))])"
)
self.assertEqual(ast.dump(node, include_attributes=True),
"Module(body=[Expr(value=Call(func=Name(id='spam', ctx=Load(), "
"lineno=1, col_offset=0), args=[Name(id='eggs', ctx=Load(), "
"lineno=1, col_offset=5), Str(s='and cheese', lineno=1, "
"col_offset=11)], keywords=[], starargs=None, kwargs=None, "
"lineno=1, col_offset=0), lineno=1, col_offset=0)])"
)
def test_classdef(self):
def cls(bases=None, keywords=None, starargs=None, kwargs=None,
body=None, decorator_list=None):
if bases is None:
bases = []
if keywords is None:
keywords = []
if body is None:
body = [ast.Pass()]
if decorator_list is None:
decorator_list = []
return ast.ClassDef("myclass", bases, keywords, starargs,
kwargs, body, decorator_list)
self.stmt(cls(bases=[ast.Name("x", ast.Store())]),
"must have Load context")
self.stmt(cls(keywords=[ast.keyword("x", ast.Name("x", ast.Store()))]),
"must have Load context")
self.stmt(cls(starargs=ast.Name("x", ast.Store())),
"must have Load context")
self.stmt(cls(kwargs=ast.Name("x", ast.Store())),
"must have Load context")
self.stmt(cls(body=[]), "empty body on ClassDef")
self.stmt(cls(body=[None]), "None disallowed")
self.stmt(cls(decorator_list=[ast.Name("x", ast.Store())]),
"must have Load context")
def test_try(self):
p = ast.Pass()
t = ast.Try([], [], [], [p])
self.stmt(t, "empty body on Try")
t = ast.Try([ast.Expr(ast.Name("x", ast.Store()))], [], [], [p])
self.stmt(t, "must have Load context")
t = ast.Try([p], [], [], [])
self.stmt(t, "Try has neither except handlers nor finalbody")
t = ast.Try([p], [], [p], [p])
self.stmt(t, "Try has orelse but no except handlers")
t = ast.Try([p], [ast.ExceptHandler(None, "x", [])], [], [])
self.stmt(t, "empty body on ExceptHandler")
e = [ast.ExceptHandler(ast.Name("x", ast.Store()), "y", [p])]
self.stmt(ast.Try([p], e, [], []), "must have Load context")
e = [ast.ExceptHandler(None, "x", [p])]
t = ast.Try([p], e, [ast.Expr(ast.Name("x", ast.Store()))], [p])
self.stmt(t, "must have Load context")
t = ast.Try([p], e, [p], [ast.Expr(ast.Name("x", ast.Store()))])
self.stmt(t, "must have Load context")
def test_subscript(self):
sub = ast.Subscript(ast.Name("x", ast.Store()), ast.Index(ast.Num(3)),
ast.Load())
self.expr(sub, "must have Load context")
x = ast.Name("x", ast.Load())
sub = ast.Subscript(x, ast.Index(ast.Name("y", ast.Store())),
ast.Load())
self.expr(sub, "must have Load context")
s = ast.Name("x", ast.Store())
for args in (s, None, None), (None, s, None), (None, None, s):
sl = ast.Slice(*args)
self.expr(ast.Subscript(x, sl, ast.Load()),
"must have Load context")
sl = ast.ExtSlice([])
self.expr(ast.Subscript(x, sl, ast.Load()), "empty dims on ExtSlice")
sl = ast.ExtSlice([ast.Index(s)])
self.expr(ast.Subscript(x, sl, ast.Load()), "must have Load context")
def pop_format_context(self, expl_expr):
"""Format the %-formatted string with current format context.
The expl_expr should be an ast.Str instance constructed from
the %-placeholders created by .explanation_param(). This will
add the required code to format said string to .on_failure and
return the ast.Name instance of the formatted string.
"""
current = self.stack.pop()
if self.stack:
self.explanation_specifiers = self.stack[-1]
keys = [ast.Str(key) for key in current.keys()]
format_dict = ast.Dict(keys, list(current.values()))
form = ast.BinOp(expl_expr, ast.Mod(), format_dict)
name = "@py_format" + str(next(self.variable_counter))
self.on_failure.append(ast.Assign([ast.Name(name, ast.Store())], form))
return ast.Name(name, ast.Load())
def specialize_constant(node, value):
if value is None or isinstance(value, bool):
new_node = ast.NameConstant(value=value)
elif isinstance(value, (int, float, complex)):
new_node = ast.Num(n=value)
elif isinstance(value, str):
new_node = ast.Str(s=value)
elif isinstance(value, bytes):
new_node = ast.Bytes(s=value)
elif isinstance(value, tuple):
elts = [specialize_constant(node, elt) for elt in value]
new_node = ast.Tuple(elts=elts, ctx=ast.Load())
else:
raise ValueError("unknown constant: %r" % value)
fatoptimizer.tools.copy_lineno(node, new_node)
return new_node
def visit_Name(self, node):
if not self.config.constant_propagation:
return
if not isinstance(node, ast.Name) or not isinstance(node.ctx, ast.Load):
return
name = node.id
if name not in self.local_variables:
# the Namespace object only tracks local variables
return
value = self.namespace.get(name)
if value is UNSET:
return
new_node = self.new_constant(node, value)
if new_node is None:
return
self.log(node, "constant propagation: replace %s with %s",
node.id, compact_ascii(value), add_line=True)
return new_node
expr.py 文件源码
项目：PyDataLondon29-EmbarrassinglyParallelDAWithAWSLambda
作者: SignalMedia
项目源码
文件源码
阅读 28
收藏 0
点赞 0
评论 0
def visit_Attribute(self, node, **kwargs):
attr = node.attr
value = node.value
ctx = node.ctx
if isinstance(ctx, ast.Load):
# resolve the value
resolved = self.visit(value).value
try:
v = getattr(resolved, attr)
name = self.env.add_tmp(v)
return self.term_type(name, self.env)
except AttributeError:
# something like datetime.datetime where scope is overridden
if isinstance(value, ast.Name) and value.id == attr:
return resolved
raise ValueError("Invalid Attribute context {0}".format(ctx.__name__))
pytables.py 文件源码
项目：PyDataLondon29-EmbarrassinglyParallelDAWithAWSLambda
作者: SignalMedia
项目源码
文件源码
阅读 24
收藏 0
点赞 0
评论 0
def visit_Attribute(self, node, **kwargs):
attr = node.attr
value = node.value
ctx = node.ctx.__class__
if ctx == ast.Load:
# resolve the value
resolved = self.visit(value)
# try to get the value to see if we are another expression
try:
resolved = resolved.value
except (AttributeError):
pass
try:
return self.term_type(getattr(resolved, attr), self.env)
except AttributeError:
# something like datetime.datetime where scope is overriden
if isinstance(value, ast.Name) and value.id == attr:
return resolved
raise ValueError("Invalid Attribute context {0}".format(ctx.__name__))
def to_source_any(n):
"""
Convert AST node to string, handling all node types, without fixing comments.
"""
try:
return astor.to_source(n)
except AttributeError:
pass
cls = n.__class__
if cls in astor.misc.all_symbols:
return astor.misc.all_symbols[cls]
def wrap(s):
return '___' + s + '___'
extra_d = {ast.Load: wrap('load'),
ast.Store: wrap('store'),
ast.Del: wrap('del'),
ast.AugLoad: wrap('augload'),
ast.AugStore: wrap('augstore'),
ast.Param: wrap('param'),
ast.keyword: wrap('keyword')}
if cls in extra_d:
return extra_d[cls]
raise AttributeError('unknown node type {}'.format(cls))
def make_subscript(varname, idx, ctx=None, lineno=0, col_offset=0):
ctx = ctx or ast.Load()
return ast.Subscript(
value = ast.Name(
id = varname,
ctx = ast.Load(),
lineno = lineno,
col_offset = col_offset,
),
slice = ast.Index(
value = ast.Num(
n = idx,
lineno = lineno,
col_offset = col_offset,
),
lineno = lineno,
col_offset = col_offset,
),
ctx = ctx,
lineno = lineno,
col_offset = col_offset,
)
def NAME(self, node):
"""
Handle occurrence of Name (which can be a load/store/delete access.)
"""
# Locate the name in locals / function / globals scopes.
if isinstance(node.ctx, (ast.Load, ast.AugLoad)):
self.handleNodeLoad(node)
if (node.id == 'locals' and isinstance(self.scope, FunctionScope)
and isinstance(node.parent, ast.Call)):
# we are doing locals() call in current scope
self.scope.usesLocals = True
elif isinstance(node.ctx, (ast.Store, ast.AugStore)):
self.handleNodeStore(node)
elif isinstance(node.ctx, ast.Del):
self.handleNodeDelete(node)
else:
# must be a Param context -- this only happens for names in function
# arguments, but these aren't dispatched through here
raise RuntimeError("Got impossible expression context: %r" % (node.ctx,))
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 test_dump(self):
node = ast.parse('spam(eggs, "and cheese")')
self.assertEqual(ast.dump(node),
"Module(body=[Expr(value=Call(func=Name(id='spam', ctx=Load()), "
"args=[Name(id='eggs', ctx=Load()), Str(s='and cheese')], "
"keywords=[], starargs=None, kwargs=None))])"
)
self.assertEqual(ast.dump(node, annotate_fields=False),
"Module([Expr(Call(Name('spam', Load()), [Name('eggs', Load()), "
"Str('and cheese')], [], None, None))])"
)
self.assertEqual(ast.dump(node, include_attributes=True),
"Module(body=[Expr(value=Call(func=Name(id='spam', ctx=Load(), "
"lineno=1, col_offset=0), args=[Name(id='eggs', ctx=Load(), "
"lineno=1, col_offset=5), Str(s='and cheese', lineno=1, "
"col_offset=11)], keywords=[], starargs=None, kwargs=None, "
"lineno=1, col_offset=0), lineno=1, col_offset=0)])"
)
def test_dump(self):
node = ast.parse('spam(eggs, "and cheese")')
self.assertEqual(ast.dump(node),
"Module(body=[Expr(value=Call(func=Name(id='spam', ctx=Load()), "
"args=[Name(id='eggs', ctx=Load()), Str(s='and cheese')], "
"keywords=[], starargs=None, kwargs=None))])"
)
self.assertEqual(ast.dump(node, annotate_fields=False),
"Module([Expr(Call(Name('spam', Load()), [Name('eggs', Load()), "
"Str('and cheese')], [], None, None))])"
)
self.assertEqual(ast.dump(node, include_attributes=True),
"Module(body=[Expr(value=Call(func=Name(id='spam', ctx=Load(), "
"lineno=1, col_offset=0), args=[Name(id='eggs', ctx=Load(), "
"lineno=1, col_offset=5), Str(s='and cheese', lineno=1, "
"col_offset=11)], keywords=[], starargs=None, kwargs=None, "
"lineno=1, col_offset=0), lineno=1, col_offset=0)])"
)
def test_classdef(self):
def cls(bases=None, keywords=None, starargs=None, kwargs=None,
body=None, decorator_list=None):
if bases is None:
bases = []
if keywords is None:
keywords = []
if body is None:
body = [ast.Pass()]
if decorator_list is None:
decorator_list = []
return ast.ClassDef("myclass", bases, keywords, starargs,
kwargs, body, decorator_list)
self.stmt(cls(bases=[ast.Name("x", ast.Store())]),
"must have Load context")
self.stmt(cls(keywords=[ast.keyword("x", ast.Name("x", ast.Store()))]),
"must have Load context")
self.stmt(cls(starargs=ast.Name("x", ast.Store())),
"must have Load context")
self.stmt(cls(kwargs=ast.Name("x", ast.Store())),
"must have Load context")
self.stmt(cls(body=[]), "empty body on ClassDef")
self.stmt(cls(body=[None]), "None disallowed")
self.stmt(cls(decorator_list=[ast.Name("x", ast.Store())]),
"must have Load context")
def test_try(self):
p = ast.Pass()
t = ast.Try([], [], [], [p])
self.stmt(t, "empty body on Try")
t = ast.Try([ast.Expr(ast.Name("x", ast.Store()))], [], [], [p])
self.stmt(t, "must have Load context")
t = ast.Try([p], [], [], [])
self.stmt(t, "Try has neither except handlers nor finalbody")
t = ast.Try([p], [], [p], [p])
self.stmt(t, "Try has orelse but no except handlers")
t = ast.Try([p], [ast.ExceptHandler(None, "x", [])], [], [])
self.stmt(t, "empty body on ExceptHandler")
e = [ast.ExceptHandler(ast.Name("x", ast.Store()), "y", [p])]
self.stmt(ast.Try([p], e, [], []), "must have Load context")
e = [ast.ExceptHandler(None, "x", [p])]
t = ast.Try([p], e, [ast.Expr(ast.Name("x", ast.Store()))], [p])
self.stmt(t, "must have Load context")
t = ast.Try([p], e, [p], [ast.Expr(ast.Name("x", ast.Store()))])
self.stmt(t, "must have Load context")
def test_subscript(self):
sub = ast.Subscript(ast.Name("x", ast.Store()), ast.Index(ast.Num(3)),
ast.Load())
self.expr(sub, "must have Load context")
x = ast.Name("x", ast.Load())
sub = ast.Subscript(x, ast.Index(ast.Name("y", ast.Store())),
ast.Load())
self.expr(sub, "must have Load context")
s = ast.Name("x", ast.Store())
for args in (s, None, None), (None, s, None), (None, None, s):
sl = ast.Slice(*args)
self.expr(ast.Subscript(x, sl, ast.Load()),
"must have Load context")
sl = ast.ExtSlice([])
self.expr(ast.Subscript(x, sl, ast.Load()), "empty dims on ExtSlice")
sl = ast.ExtSlice([ast.Index(s)])
self.expr(ast.Subscript(x, sl, ast.Load()), "must have Load context")
def make_if(kwd, test, then, *rest):
# (This'd be simpler with a different form of the grammar.)
test = test(ast.Load())
if not rest: else_ = []
elif len(rest) == 1: else_ = rest[0]
else: else_ = [make_if(*rest)]
return ast.If(test, then, else_,
lineno=kwd.start[0],
col_offset=kwd.start[1])
def maybe_assignment(*expr_fns):
if len(expr_fns) == 1:
node0 = expr_fns[0](ast.Load())
stmt = ast.Expr(node0)
else:
lhses = [fn(ast.Store()) for fn in expr_fns[:-1]]
node0 = lhses[0]
stmt = ast.Assign(lhses, expr_fns[-1](ast.Load()))
return ast.copy_location(stmt, node0)
def visit_Attribute(self, t):
self(t.value)
self.check_identifier(t.attr)
if isinstance(t.ctx, ast.Load): pass
elif isinstance(t.ctx, ast.Store): pass
else: assert False, "Only loads and stores are supported: %r" % (t,)
def visit_Subscript(self, t):
self(t.value)
if isinstance(t.slice, ast.Index):
if isinstance(t.ctx, ast.Load): pass
elif isinstance(t.ctx, ast.Store): pass
else: assert False, "Only loads and stores are supported: %r" % (t,)
self(t.slice.value)
else:
assert False, "Only simple subscripts are supported: %r" % (t,)
