use basemodel for structures

sdk/python/kfp/components/base_component.py

@@ -74,7 +74,7 @@ def __call__(self, *args, **kwargs) -> pipeline_task.PipelineTask:
         missing_arguments = [
             input_name for input_name, input_spec in (
                 self.component_spec.inputs or {}).items()
-            if input_name not in task_inputs and not input_spec.optional and
+            if input_name not in task_inputs and not input_spec._optional and
             not type_utils.is_task_final_status_type(input_spec.type)
         ]
         if missing_arguments:

sdk/python/kfp/components/pipeline_task.py

@@ -13,13 +13,13 @@
 # limitations under the License.
 """Pipeline task class and operations."""
 
-import re
 import copy
+import re
 from typing import Any, List, Mapping, Optional, Union
 
 from kfp.components import constants
-from kfp.components import placeholders
 from kfp.components import pipeline_channel
+from kfp.components import placeholders
 from kfp.components import structures
 from kfp.components.types import type_utils
 
@@ -116,6 +116,7 @@ def __init__(
         self.container_spec = None
 
         if component_spec.implementation.container is not None:
+
             self.container_spec = self._resolve_command_line_and_arguments(
                 component_spec=component_spec,
                 args=args,
@@ -266,7 +267,7 @@ def expand_command_part(arg) -> Union[str, List[str], None]:
                     return input_path
                 else:
                     input_spec = inputs_dict[input_name]
-                    if input_spec.optional:
+                    if input_spec._optional:
                         return None
                     else:
                         raise ValueError(