Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
self.__compare_obj_list(getattr(obj, snake_key), value, obj_type, param_name)
else:
# Straight list comparison
self.__compare_list(getattr(obj, snake_key), value, param_name)
elif item_kind and item_kind.startswith('dict('):
self.__compare_dict(getattr(obj, snake_key), value, param_name)
elif item_kind and type(value).__name__ == 'dict':
# object
param_obj = getattr(obj, snake_key)
if not param_obj:
setattr(obj, snake_key, self.model_class_from_name(item_kind)())
param_obj = getattr(obj, snake_key)
self.__update_object_properties(param_obj, value)
else:
if item_kind:
raise KubernetesException(
"Evaluating {0}: encountered unimplemented type {1} in "
"__compare_obj_list() for model {2}".format(
param_name,
item_kind,
self.get_base_model_name_snake(obj_class))
)
else:
raise KubernetesException(
"Evaluating {0}: unable to get swagger_type for {1} in "
"__compare_obj_list() for item {2} in model {3}".format(
param_name,
snake_key,
str(item),
self.get_base_model_name_snake(obj_class))
)
if not found:
:type message: basestring
:param kwargs: All other keyword arguments.
:type kwargs: dict
"""
self.message = message
self.value = kwargs
self.value['message'] = message
def __str__(self):
"""
String representation of the instance.
"""
return json.dumps(self.value)
class OpenShiftException(KubernetesException):
pass
:param param_value: The value to set.
:return: The original object.
"""
while len(property_path) > 0:
raw_prop_name = property_path.pop(0)
prop_name = PYTHON_KEYWORD_MAPPING.get(raw_prop_name, raw_prop_name)
prop_kind = obj.swagger_types[prop_name]
if prop_kind in PRIMITIVES:
try:
setattr(obj, prop_name, param_value)
except ValueError as exc:
msg = str(exc)
if param_value is None and 'None' in msg:
pass
else:
raise KubernetesException(
"Error setting {0} to {1}: {2}".format(prop_name, param_value, msg)
)
elif prop_kind.startswith('dict('):
if not getattr(obj, prop_name):
setattr(obj, prop_name, param_value)
else:
self.__compare_dict(getattr(obj, prop_name), param_value, param_name)
elif prop_kind.startswith('list['):
if getattr(obj, prop_name) is None:
setattr(obj, prop_name, [])
obj_type = prop_kind.replace('list[', '').replace(']', '')
if obj_type not in PRIMITIVES and obj_type not in ('list', 'dict'):
self.__compare_obj_list(getattr(obj, prop_name), param_value, obj_type, param_name)
else:
self.__compare_list(getattr(obj, prop_name), param_value, param_name)
else:
def find_arg_spec(self, module_param_name):
"""For testing, allow the param_name value to be an alias"""
if module_param_name in self.argspec:
return self.argspec[module_param_name]
result = None
for key, value in iteritems(self.argspec):
if value.get('aliases'):
for alias in value['aliases']:
if alias == module_param_name:
result = self.argspec[key]
break
if result:
break
if not result:
raise KubernetesException(
"Error: received unrecognized module parameter {0}".format(module_param_name)
)
return result
:param param_value: The value to set.
:return: The original object.
"""
while len(property_path) > 0:
raw_prop_name = property_path.pop(0)
prop_name = PYTHON_KEYWORD_MAPPING.get(raw_prop_name, raw_prop_name)
prop_kind = obj.swagger_types[prop_name]
if prop_kind in PRIMITIVES:
try:
setattr(obj, prop_name, param_value)
except ValueError as exc:
msg = str(exc)
if param_value is None and 'None' in msg:
pass
else:
raise KubernetesException(
"Error setting {0} to {1}: {2}".format(prop_name, param_value, msg)
)
elif prop_kind.startswith('dict('):
if not getattr(obj, prop_name):
setattr(obj, prop_name, param_value)
else:
self.__compare_dict(getattr(obj, prop_name), param_value, param_name)
elif prop_kind.startswith('list['):
if getattr(obj, prop_name) is None:
setattr(obj, prop_name, [])
obj_type = prop_kind.replace('list[', '').replace(']', '')
if obj_type not in PRIMITIVES and obj_type not in ('list', 'dict'):
self.__compare_obj_list(getattr(obj, prop_name), param_value, obj_type, param_name)
else:
self.__compare_list(getattr(obj, prop_name), param_value, param_name)
else:
def run_docstrings_cmd(**kwargs):
"""
Send documentation string and return string to stdout for each model requested
:param kwargs: parser arguments
:return: None
"""
models = kwargs.pop('models')
api_version = kwargs.pop('api_version')
if not models:
raise KubernetesException(
"ERROR: you must provide one or more models for the docstrings command."
)
for model in models:
try:
strings = KubernetesDocStrings(model=model, api_version=api_version)
except KubernetesException:
try:
strings = OpenShiftDocStrings(model=model, api_version=api_version)
except KubernetesException:
raise
print("DOCUMENTATION = '''")
print(strings.documentation)
print("'''\n")
print("RETURN = '''")
print(strings.return_block)
def __update_object_properties(self, obj, item):
""" Recursively update an object's properties. Returns a pointer to the object. """
for key, value in iteritems(item):
snake_key = self.attribute_to_snake(key)
try:
kind = obj.swagger_types[snake_key]
except (AttributeError, KeyError):
possible_matches = ', '.join(list(obj.swagger_types.keys()))
class_snake_name = self.get_base_model_name_snake(type(obj).__name__)
raise KubernetesException(
"Unable to find '{0}' in {1}. Valid property names include: {2}".format(snake_key,
class_snake_name,
possible_matches)
)
if kind in PRIMITIVES or kind.startswith('list[') or kind.startswith('dict('):
self.__set_obj_attribute(obj, [snake_key], value, snake_key)
else:
# kind is an object, hopefully
if not getattr(obj, snake_key):
setattr(obj, snake_key, self.model_class_from_name(kind)())
self.__update_object_properties(getattr(obj, snake_key), value)
return obj
if not match:
missing.append(request_dict)
src_values += missing
elif type(src_values[0]).__name__ == 'list':
missing = []
for request_list in request_values:
match = False
for src_list in src_values:
if set(request_list) >= set(src_list):
match = True
break
if not match:
missing.append(request_list)
src_values += missing
else:
raise KubernetesException(
"Evaluating {0}: encountered unimplemented type {1} in "
"__compare_list()".format(param_name, type(src_values[0]).__name__)
)
def find_arg_spec(self, module_param_name):
"""For testing, allow the param_name value to be an alias"""
if module_param_name in self.argspec:
return self.argspec[module_param_name]
result = None
for key, value in iteritems(self.argspec):
if value.get('aliases'):
for alias in value['aliases']:
if alias == module_param_name:
result = self.argspec[key]
break
if result:
break
if not result:
raise KubernetesException(
"Error: received unrecognized module parameter {0}".format(module_param_name)
)
return result
:param param_value: The value to set.
:return: The original object.
"""
while len(property_path) > 0:
raw_prop_name = property_path.pop(0)
prop_name = PYTHON_KEYWORD_MAPPING.get(raw_prop_name, raw_prop_name)
prop_kind = obj.swagger_types[prop_name]
if prop_kind in PRIMITIVES:
try:
setattr(obj, prop_name, param_value)
except ValueError as exc:
msg = str(exc)
if param_value is None and 'None' in msg:
pass
else:
raise KubernetesException(
"Error setting {0} to {1}: {2}".format(prop_name, param_value, msg)
)
elif prop_kind.startswith('dict('):
if not getattr(obj, prop_name):
setattr(obj, prop_name, param_value)
else:
self.__compare_dict(getattr(obj, prop_name), param_value, param_name)
elif prop_kind.startswith('list['):
if getattr(obj, prop_name) is None:
setattr(obj, prop_name, [])
obj_type = prop_kind.replace('list[', '').replace(']', '')
if obj_type not in PRIMITIVES and obj_type not in ('list', 'dict'):
self.__compare_obj_list(getattr(obj, prop_name), param_value, obj_type, param_name)
else:
self.__compare_list(getattr(obj, prop_name), param_value, param_name)
else: