SDK - Made ComponentSpec.implementation field optional (#1188)

* SDK - Made ComponentSpec.implementation field optional
Improved the error message when trying to convert tasks to ContainerOp.

* Switched from attribute checking to type checking

sdk/python/kfp/components/_dsl_bridge.py

@@ -14,16 +14,16 @@
 
 from collections import OrderedDict
 from typing import Mapping
-from ._structures import ConcatPlaceholder, IfPlaceholder, InputValuePlaceholder, InputPathPlaceholder, IsPresentPlaceholder, OutputPathPlaceholder, TaskSpec
+from ._structures import ContainerImplementation, ConcatPlaceholder, IfPlaceholder, InputValuePlaceholder, InputPathPlaceholder, IsPresentPlaceholder, OutputPathPlaceholder, TaskSpec
 from ._components import _generate_output_file_name, _default_component_name
 from kfp.dsl._metadata import ComponentMeta, ParameterMeta, TypeMeta, _annotation_to_typemeta
 
 def create_container_op_from_task(task_spec: TaskSpec):
     argument_values = task_spec.arguments
     component_spec = task_spec.component_ref._component_spec
 
-    if hasattr(component_spec.implementation, 'graph'):
-        raise TypeError('Cannot convert graph component to ContainerOp')
+    if not isinstance(component_spec.implementation, ContainerImplementation):
+        raise TypeError('Only container component tasks can be converted to ContainerOp')
 
     inputs_dict = {input_spec.name: input_spec for input_spec in component_spec.inputs or []}
     container_spec = component_spec.implementation.container

sdk/python/kfp/components/_structures.py

@@ -227,12 +227,12 @@ class ComponentSpec(ModelBase):
     '''Component specification. Describes the metadata (name, description, annotations and labels), the interface (inputs and outputs) and the implementation of the component.'''
     def __init__(
         self,
-        implementation: ImplementationType,
         name: Optional[str] = None, #? Move to metadata?
         description: Optional[str] = None, #? Move to metadata?
         metadata: Optional[MetadataSpec] = None,
         inputs: Optional[List[InputSpec]] = None,
         outputs: Optional[List[OutputSpec]] = None,
+        implementation: Optional[ImplementationType] = None,
         version: Optional[str] = 'google.com/cloud/pipelines/component/v1',
         #tags: Optional[Set[str]] = None,
     ):