Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_none(self):
""" what to do when names isn't defined, or is 'none' """
with self.assertRaises(NameNotDefined):
self.t("a == 2", None)
self.s.names["s"] = 21
with self.assertRaises(NameNotDefined):
self.t("s += a", 21)
self.s.names = None
with self.assertRaises(InvalidExpression):
self.t('s', 21)
self.s.names = {'a': {'b': {'c': 42}}}
with self.assertRaises(AttributeDoesNotExist):
self.t('a.b.d**2', 42)
# pylint: disable=bad-super-call
super(InvalidExpression, self).__init__(self.message)
class AttributeDoesNotExist(InvalidExpression):
"""attribute does not exist"""
def __init__(self, attr, expression):
self.message = \
"Attribute '{0}' does not exist in expression '{1}'".format(
attr, expression)
self.attr = attr
self.expression = expression
class FeatureNotAvailable(InvalidExpression):
""" What you're trying to do is not allowed. """
pass
class NumberTooHigh(InvalidExpression):
""" Sorry! That number is too high. I don't want to spend the
next 10 years evaluating this expression! """
pass
class IterableTooLong(InvalidExpression):
""" That iterable is **way** too long, baby. """
pass
class FunctionNotDefined(InvalidExpression):
""" sorry! That function isn't defined! """
def __init__(self, func_name, expression):
self.message = "Function '{0}' not defined," \
" for expression '{1}'.".format(func_name, expression)
setattr(self, 'func_name', func_name) # bypass 2to3 confusion.
self.expression = expression
# pylint: disable=bad-super-call
super(InvalidExpression, self).__init__(self.message)
class NameNotDefined(InvalidExpression):
""" a name isn't defined. """
def __init__(self, name, expression):
self.name = name
self.message = "'{0}' is not defined for expression '{1}'".format(
name, expression)
self.expression = expression
# pylint: disable=bad-super-call
super(InvalidExpression, self).__init__(self.message)
class AttributeDoesNotExist(InvalidExpression):
"""attribute does not exist"""
def __init__(self, attr, expression):
def __call__(self, item, context=None):
var_dict = self.get_variables(item, context)
try:
untransformed_value = eval_statements(self.statement, var_dict)
return transform_from_datatype(self.datatype)(untransformed_value)
except (InvalidExpression, SyntaxError, TypeError, ZeroDivisionError):
return None
class NameNotDefined(InvalidExpression):
""" a name isn't defined. """
def __init__(self, name, expression):
self.name = name
self.message = "'{0}' is not defined for expression '{1}'".format(
name, expression)
self.expression = expression
# pylint: disable=bad-super-call
super(InvalidExpression, self).__init__(self.message)
class AttributeDoesNotExist(InvalidExpression):
"""attribute does not exist"""
def __init__(self, attr, expression):
self.message = \
"Attribute '{0}' does not exist in expression '{1}'".format(
attr, expression)
self.attr = attr
self.expression = expression
class FeatureNotAvailable(InvalidExpression):
""" What you're trying to do is not allowed. """
pass
def __init__(self, name, expression):
self.name = name
self.message = "'{0}' is not defined for expression '{1}'".format(
name, expression)
self.expression = expression
# pylint: disable=bad-super-call
super(InvalidExpression, self).__init__(self.message)
def populate_params_runtime_values(self):
inline.open_session()
try:
global_context = {}
for param in Param.where(pipeline_id=None, job_id=None).all():
global_context[param.name] = param.populate_runtime_value()
pipeline_context = global_context.copy()
for param in self.params.all():
pipeline_context[param.name] = param.populate_runtime_value(global_context)
for job in self.jobs.all():
for param in job.params.all():
param.populate_runtime_value(pipeline_context)
inline.close_session()
return True
except (InvalidExpression, TypeError, ValueError, SyntaxError) as e:
inline.close_session()
from core import cloud_logging
job_id = 'N/A'
worker_class = 'N/A'
if param.job_id is not None:
job_id = param.job_id
worker_class = param.job.worker_class
message = 'Invalid job parameter "%s": %s' % (param.label, e)
elif param.pipeline_id is not None:
message = 'Invalid pipeline variable "%s": %s' % (param.label, e)
else:
message = 'Invalid global variable "%s": %s' % (param.label, e)
cloud_logging.logger.log_struct({
'labels': {
'pipeline_id': self.id,
'job_id': job_id,
def eval_statements(statement, variable_context):
"""Evaluates math statements and returns the value
args
statement: a simple python-like math statement
variable_context: a dict with variable names as key and assigned values as dict values
"""
# variable values should be numbers
var_types = set(type(value) for value in variable_context.values())
if not var_types.issubset({float, Decimal, date, datetime, NoneType, bool}.union(set(six.integer_types))):
raise InvalidExpression
evaluator = SimpleEval(operators=SAFE_OPERATORS, names=variable_context, functions=FUNCTIONS)
return evaluator.eval(statement)
def parse_expr(self, expr: str):
"""
Parse an expression in string
"""
try:
v = self._evaluator.eval(expr=expr)
except SyntaxError as ex:
v = expr
# TODO: feature T2
# print("Raised simpleeval exception {} for expression {}".format(type(ex), v))
self.errors.append([expr, ex])
except simpleeval.InvalidExpression as ex:
v = expr
# TODO: feature T2
# print("Raised simpleeval exception {} for expression {}".format(type(ex), v))
# print("Raised simpleeval exception {} for expression {}".format(type(ex), v))
# print("Message: {}".format(ex))
self.errors.append(ex)
except Exception as ex:
v = expr
# TODO: feature T2
# print('Exception on expression: {}'.format(expr))
self.errors.append(ex)
raise
# TODO: feature T4: include this part of the classes so user could override
v = filter_value(v)
return v
if PYTHON3:
exec('DISALLOW_FUNCTIONS.add(exec)') # exec is not a function in Python2...
########################################
# Exceptions:
class InvalidExpression(Exception):
""" Generic Exception """
pass
class FunctionNotDefined(InvalidExpression):
""" sorry! That function isn't defined! """
def __init__(self, func_name, expression):
self.message = "Function '{0}' not defined," \
" for expression '{1}'.".format(func_name, expression)
setattr(self, 'func_name', func_name) # bypass 2to3 confusion.
self.expression = expression
# pylint: disable=bad-super-call
super(InvalidExpression, self).__init__(self.message)
class NameNotDefined(InvalidExpression):
""" a name isn't defined. """
def __init__(self, name, expression):