feat(sdk): added sanitizing k8s object process and test. Fixes: #332

sdk/python/kfp_tekton/compiler/_k8s_helper.py

@@ -115,3 +115,90 @@ def convert_k8s_obj_to_json(k8s_obj):
 
     return {key: convert_k8s_obj_to_json(val)
             for key, val in iteritems(obj_dict)}
+
+
+def _to_str(s):
+    return None if s is None else str(s)
+
+
+def _to_bool(b):
+    if type(b) == str:
+      b = b.lower()
+      if b != 'true' and b != 'false':
+        raise ValueError(
+          'Invalid boolean string {}. Should be "true" or "false".'.format(b))
+      else:
+        b = b == 'true'
+    elif type(b) != bool:
+      raise ValueError('Invalid value {}. Should be boolean.'.format(b))
+    return b
+
+
+def _to_int(i):
+    try:
+      result = int(i)
+    except ValueError:
+      raise ValueError('Invalid {}. Should be integer.'.format(i))
+    return result
+
+
+def _to_float(f):
+    try:
+      result = float(f)
+    except ValueError:
+      raise ValueError('Invalid {}. Should be float'.format(f))
+    return result
+
+
+def sanitize_k8s_object(k8s_obj, type=None):
+    """
+    Recursively sanitize and cast k8s object type based on the type definition
+    in kubernetes.client.models.
+
+    Args:
+      k8s_obj: k8s object
+    """
+    from six import text_type, integer_types
+    PRIMITIVE_TYPES = (float, bool, bytes, text_type) + integer_types
+    from datetime import date, datetime
+
+    if k8s_obj is None:
+      return None
+    elif isinstance(k8s_obj, PRIMITIVE_TYPES):
+      if type == 'str':
+        return _to_str(k8s_obj)
+      elif type == 'int':
+        return _to_int(k8s_obj)
+      elif type == 'float':
+        return _to_float(k8s_obj)
+      elif type == 'bool':
+        return _to_bool(k8s_obj)
+      else:
+        return k8s_obj
+    elif isinstance(k8s_obj, list):
+      if type == 'list[str]':
+        return [sanitize_k8s_object(sub_obj, 'str')
+                for sub_obj in k8s_obj]
+      else:
+        return [sanitize_k8s_object(sub_obj, None)
+                for sub_obj in k8s_obj]
+    elif isinstance(k8s_obj, tuple):
+      if type == 'list[str]':
+        return tuple(sanitize_k8s_object(sub_obj, 'str')
+                    for sub_obj in k8s_obj)
+      else:
+        return tuple(sanitize_k8s_object(sub_obj, None)
+                    for sub_obj in k8s_obj)
+    elif isinstance(k8s_obj, (datetime, date)):
+      return k8s_obj
+
+    if isinstance(k8s_obj, dict):
+      return k8s_obj
+    else:
+      for attr in k8s_obj.attribute_map:
+        if getattr(k8s_obj, attr) is not None:
+          type = k8s_obj.openapi_types[attr]
+          value = getattr(k8s_obj, attr)
+          value = sanitize_k8s_object(value, type)
+          setattr(k8s_obj, attr, value)
+      return k8s_obj

sdk/python/kfp_tekton/compiler/compiler.py

@@ -37,7 +37,7 @@
 # KFP-Tekton imports
 from kfp_tekton.compiler import __tekton_api_version__ as tekton_api_version
 from kfp_tekton.compiler._data_passing_rewriter import fix_big_data_passing
-from kfp_tekton.compiler._k8s_helper import convert_k8s_obj_to_json, sanitize_k8s_name
+from kfp_tekton.compiler._k8s_helper import convert_k8s_obj_to_json, sanitize_k8s_name, sanitize_k8s_object
 from kfp_tekton.compiler._op_to_template import _op_to_template
 
 
@@ -583,6 +583,8 @@ def _sanitize_and_inject_artifact(self, pipeline: dsl.Pipeline, pipeline_conf=No
           sanitized_attribute_outputs[sanitize_k8s_name(key, True)] = \
             op.attribute_outputs[key]
         op.attribute_outputs = sanitized_attribute_outputs
+      if isinstance(op, dsl.ContainerOp) and op.container is not None:
+        sanitize_k8s_object(op.container)
       sanitized_ops[sanitized_name] = op
     pipeline.ops = sanitized_ops