def check_params(self, params):
'''Check for user typos in "params" keys to avoid
unwanted usage of default values
# Arguments
params: dictionary
The parameters to be checked
'''
legal_params_fns = [Sequential.fit, Sequential.predict,
Sequential.predict_classes, Sequential.evaluate]
if self.build_fn is None:
legal_params_fns.append(self.__call__)
elif not isinstance(self.build_fn, types.FunctionType) and not isinstance(self.build_fn, types.MethodType):
legal_params_fns.append(self.build_fn.__call__)
else:
legal_params_fns.append(self.build_fn)
legal_params = []
for fn in legal_params_fns:
legal_params += inspect.getargspec(fn)[0]
legal_params = set(legal_params)
for params_name in params:
if params_name not in legal_params:
raise ValueError('{} is not a legal parameter'.format(params_name))
python类FunctionType()的实例源码
def get_undecorated_callback(self):
""" Return the callback. If the callback is a decorated function, try to
recover the original function. """
func = self.callback
func = getattr(func, '__func__' if py3k else 'im_func', func)
closure_attr = '__closure__' if py3k else 'func_closure'
while hasattr(func, closure_attr) and getattr(func, closure_attr):
attributes = getattr(func, closure_attr)
func = attributes[0].cell_contents
# in case of decorators with multiple arguments
if not isinstance(func, FunctionType):
# pick first FunctionType instance from multiple arguments
func = filter(lambda x: isinstance(x, FunctionType),
map(lambda x: x.cell_contents, attributes))
func = list(func)[0] # py3 support
return func
def obj2dict(obj, processed=None):
"""
converts any object into a dict, recursively
"""
processed = processed if not processed is None else set()
if obj is None:
return None
if isinstance(obj,(int,long,str,unicode,float,bool)):
return obj
if id(obj) in processed:
return '<reference>'
processed.add(id(obj))
if isinstance(obj,(list,tuple)):
return [obj2dict(item,processed) for item in obj]
if not isinstance(obj, dict) and hasattr(obj,'__dict__'):
obj = obj.__dict__
else:
return repr(obj)
return dict((key,obj2dict(value,processed)) for key,value in obj.items()
if not key.startswith('_') and
not type(value) in (types.FunctionType,
types.LambdaType,
types.BuiltinFunctionType,
types.BuiltinMethodType))
def _compute_fields_for_operation(self, fields, to_compute):
row = OpRow(self)
for name, tup in iteritems(fields):
field, value = tup
if isinstance(
value, (
types.LambdaType, types.FunctionType, types.MethodType,
types.BuiltinFunctionType, types.BuiltinMethodType
)
):
value = value()
row.set_value(name, value, field)
for name, field in to_compute:
try:
row.set_value(name, field.compute(row), field)
except (KeyError, AttributeError):
# error silently unless field is required!
if field.required and name not in fields:
raise RuntimeError(
'unable to compute required field: %s' % name)
return row
def __getattr__(self, name):
""" Handle unknown attribute lookups """
if name == "readonly":
return True
sname = name.split('_')
if sname[:2] == ['get', 'unfiltered']:
"""
Handle get_unfiltered calls. Return the name of the access
method for the base database object. Call setattr before
returning so that the lookup happens at most once for a given
method call and a given object.
"""
attr = getattr(self.basedb, 'get_' + sname[2] + '_from_handle')
setattr(self, name, attr)
return attr
# if a write-method:
if (name in DbWriteBase.__dict__ and
not name.startswith("__") and
type(DbWriteBase.__dict__[name]) is types.FunctionType):
raise AttributeError
# Default behaviour: lookup attribute in parent object
return getattr(self.db, name)
def __new__(cls, expectation=None, strcmp=None, is_custom_func=False):
"""
Args:
anything (expectation: the object you want to compare)
"substring", "regex" (strcmp: the type of string comparsion)
Boolean (is_custom_func: set True is object is a test function of Matcher
Return:
Matcher
"""
options = {}
if is_custom_func:
if isinstance(expectation, (FunctionType, BuiltinFunctionType, MethodType)):
options["is_custom_func"] = True
else:
# Todo: customized error exception
raise TypeError("[{}] is not callable".format(expectation))
if strcmp:
if isinstance(expectation, (str, unicode)):
if strcmp.upper() == "DEFAULT" or strcmp.upper() == "SUBSTRING":
options["is_substring"] = True
elif strcmp.upper() == "REGEX":
options["is_regex"] = True
else:
raise TypeError("[{}] is not a string".format(expectation))
return Matcher(expectation, options)
def init(scope):
"""
Copy all values of scope into the class SinonGlobals
Args:
scope (eg. locals() or globals())
Return:
SinonGlobals instance
"""
class SinonGlobals(object): #pylint: disable=too-few-public-methods
"""
A fully empty class
External can push the whole `scope` into this class through global function init()
"""
pass
global CPSCOPE #pylint: disable=global-statement
CPSCOPE = SinonGlobals()
funcs = [obj for obj in scope.values() if isinstance(obj, FunctionType)]
for func in funcs:
setattr(CPSCOPE, func.__name__, func)
return CPSCOPE
def sw_compute_features(learn_data, overwrite_existing=False, worker_id=None):
# learn_data = db['learns'].find_one(learn_id)
model_data = db[learn_data['Model'][-1]].find_one(learn_data['Model'][0])
# sample_df = load_df_from_sample_notation(model_data['Initial Sample Location'])
if not check_sample_exists(model_data['Feature Sample Location']) or overwrite_existing:
feature_generating_function_code = marshal.loads(db[model_data['Feature Generation Function'][-1]]\
.find_one(model_data['Feature Generation Function'][0])['Code'])
feature_generating_function = types.FunctionType(feature_generating_function_code, globals())
# save_df_to_sample_notation(, model_data['Feature Sample Location'])
learn_data = feature_generating_function(learn_data, model_data)
learn_data['Status']['Features Computed'] = True
db['learns'].update(learn_data['_id'], learn_data)
def api_exc_patch(func):
if isinstance(func, FunctionType):
@wraps(func)
def deco(*args, **kwargs):
try:
return func(*args, **kwargs)
except RQInvalidArgument:
raise
except Exception as e:
if isinstance(e, TypeError):
exc_info = sys.exc_info()
try:
ret = inspect.getcallargs(unwrapper(func), *args, **kwargs)
except TypeError:
t, v, tb = exc_info
raise patch_user_exc(v.with_traceback(tb))
if getattr(e, EXC_EXT_NAME, EXC_TYPE.NOTSET) == EXC_TYPE.NOTSET:
patch_system_exc(e)
raise
return deco
return func
def api_exc_patch(func):
if isinstance(func, FunctionType):
@wraps(func)
def deco(*args, **kwargs):
try:
return func(*args, **kwargs)
except RQInvalidArgument:
raise
except Exception as e:
if isinstance(e, TypeError):
exc_info = sys.exc_info()
try:
ret = inspect.getcallargs(unwrapper(func), *args, **kwargs)
except TypeError:
t, v, tb = exc_info
raise patch_user_exc(v.with_traceback(tb))
if getattr(e, EXC_EXT_NAME, EXC_TYPE.NOTSET) == EXC_TYPE.NOTSET:
patch_system_exc(e)
raise
return deco
return func
def extract_translation_type(obj):
result = extract_translation.dispatch(object)(obj)
for attr in dir(obj):
if attr.startswith('_'):
continue
method = getattr(obj, attr, None)
if isinstance(method, types.FunctionType):
trans = extract_translation(method)
for k, v in trans.items():
result[':%s.%s' % (attr, k)] = v
else:
result[':%s' % attr] = attr
return result
def validate_error_func(self):
if self.error_func:
if isinstance(self.error_func, types.FunctionType):
ismethod = 0
elif isinstance(self.error_func, types.MethodType):
ismethod = 1
else:
self.log.error("'p_error' defined, but is not a function or method")
self.error = True
return
eline = self.error_func.__code__.co_firstlineno
efile = self.error_func.__code__.co_filename
module = inspect.getmodule(self.error_func)
self.modules.add(module)
argcount = self.error_func.__code__.co_argcount - ismethod
if argcount != 1:
self.log.error('%s:%d: p_error() requires 1 argument', efile, eline)
self.error = True
# Get the tokens map
def get_pfunctions(self):
p_functions = []
for name, item in self.pdict.items():
if not name.startswith('p_') or name == 'p_error':
continue
if isinstance(item, (types.FunctionType, types.MethodType)):
line = item.__code__.co_firstlineno
module = inspect.getmodule(item)
p_functions.append((line, module, name, item.__doc__))
# Sort all of the actions by line number; make sure to stringify
# modules to make them sortable, since `line` may not uniquely sort all
# p functions
p_functions.sort(key=lambda p_function: (
p_function[0],
str(p_function[1]),
p_function[2],
p_function[3]))
self.pfuncs = p_functions
# Validate all of the p_functions
def typeof(self, cdecl):
"""Parse the C type given as a string and return the
corresponding <ctype> object.
It can also be used on 'cdata' instance to get its C type.
"""
if isinstance(cdecl, basestring):
return self._typeof(cdecl)
if isinstance(cdecl, self.CData):
return self._backend.typeof(cdecl)
if isinstance(cdecl, types.BuiltinFunctionType):
res = _builtin_function_type(cdecl)
if res is not None:
return res
if (isinstance(cdecl, types.FunctionType)
and hasattr(cdecl, '_cffi_base_type')):
with self._lock:
return self._get_cached_btype(cdecl._cffi_base_type)
raise TypeError(type(cdecl))
def validate_error_func(self):
if self.error_func:
if isinstance(self.error_func,types.FunctionType):
ismethod = 0
elif isinstance(self.error_func, types.MethodType):
ismethod = 1
else:
self.log.error("'p_error' defined, but is not a function or method")
self.error = 1
return
eline = func_code(self.error_func).co_firstlineno
efile = func_code(self.error_func).co_filename
self.files[efile] = 1
if (func_code(self.error_func).co_argcount != 1+ismethod):
self.log.error("%s:%d: p_error() requires 1 argument",efile,eline)
self.error = 1
# Get the tokens map
def get_pfunctions(self):
p_functions = []
for name, item in self.pdict.items():
if name[:2] != 'p_': continue
if name == 'p_error': continue
if isinstance(item,(types.FunctionType,types.MethodType)):
line = func_code(item).co_firstlineno
file = func_code(item).co_filename
p_functions.append((line,file,name,item.__doc__))
# Sort all of the actions by line number
p_functions.sort()
self.pfuncs = p_functions
# Validate all of the p_functions
def register(cls, action):
action_logger.info("Registering action :%s" % action)
if isinstance(action, (types.FunctionType, staticmethod)):
name = action.__name__
cls._actions[name.upper()] = action
setattr(cls, name, action)
elif isinstance(action, types.ClassType) and hasattr(action, "__call__"):
name = action.__name__
action = action()
cls._actions[name.upper()] = action
setattr(cls, name, action)
elif (isinstance(action, (types.InstanceType, types.ObjectType))
and hasattr(action, "__call__")):
if isinstance(action, type):
name = action.__name__
action = action()
else:
name = action.__class__.__name__
cls._actions[name.upper()] = action
setattr(cls, name, action)
else:
name = str(action)
action_logger.error("Error registering action :%s" % action)
raise UnknownAction("unable to register action %s!!" %name)
def register(cls, event):
exe_logger.info("Registering event :%s" % event)
if isinstance(event, (types.FunctionType, staticmethod)):
name = event.__name__
cls._chaos_events[name.upper()] = event
setattr(cls, name, event)
elif isinstance(event, types.ClassType) and hasattr(event, "__call__"):
name = event.__name__
event = event()
cls._chaos_events[name.upper()] = event
setattr(cls, name, event)
elif (isinstance(event, (types.InstanceType, types.ObjectType))
and hasattr(event, "__call__")):
if isinstance(event, type):
name = event.__name__
event = event()
else:
name = event.__class__.__name__
cls._chaos_events[name.upper()] = event
setattr(cls, name, event)
else:
name = str(event)
exe_logger.error("Error registering event :%s" % event)
raise UnknownChaosEvent("unable to register event %s!!" % name)
def method_decorator_metaclass(function):
class Decorated(type):
def __new__(meta, classname, bases, classDict):
for attributeName, attribute in classDict.items():
if isinstance(attribute, types.FunctionType):
attribute = function(attribute)
classDict[attributeName] = attribute
return type.__new__(meta, classname, bases, classDict)
return Decorated
def get_unpatched(item):
lookup = (
get_unpatched_class if isinstance(item, six.class_types) else
get_unpatched_function if isinstance(item, types.FunctionType) else
lambda item: None
)
return lookup(item)
