lint

Signed-off-by: Kevin Su <pingsutw@apache.org>

flytekit/core/base_task.py

@@ -340,8 +340,8 @@ def sandbox_execute(
         """
         Call dispatch_execute, in the context of a local sandbox execution. Not invoked during runtime.
         """
-        es = ctx.execution_state
-        b = es.user_space_params.with_task_sandbox()
+        es = cast(ExecutionState, ctx.execution_state)
+        b = cast(ExecutionParameters, es.user_space_params).with_task_sandbox()
         ctx = ctx.current_context().with_execution_state(es.with_params(user_space_params=b.build())).build()
         return self.dispatch_execute(ctx, input_literal_map)
 

flytekit/core/python_auto_container.py

@@ -3,7 +3,7 @@
 import importlib
 import re
 from abc import ABC
-from typing import Any, Callable, Dict, List, Optional, TypeVar, Union
+from typing import Any, Callable, Dict, List, Optional, TypeVar, cast
 
 from flyteidl.core import tasks_pb2 as _core_task
 from kubernetes.client import ApiClient
@@ -207,23 +207,23 @@ def _get_container(self, settings: SerializationSettings) -> _task_model.Contain
         )
 
     def _serialize_pod_spec(self, settings: SerializationSettings) -> Dict[str, Any]:
-        containers = self.pod_template.pod_spec.containers
+        containers = cast(PodTemplate, self.pod_template).pod_spec.containers
         primary_exists = False
 
         for container in containers:
-            if container.name == self.pod_template.primary_container_name:
+            if container.name == cast(PodTemplate, self.pod_template).primary_container_name:
                 primary_exists = True
                 break
 
         if not primary_exists:
             # insert a placeholder primary container if it is not defined in the pod spec.
-            containers.append(V1Container(name=self.pod_template.primary_container_name))
+            containers.append(V1Container(name=cast(PodTemplate, self.pod_template).primary_container_name))
         final_containers = []
         for container in containers:
             # In the case of the primary container, we overwrite specific container attributes
             # with the default values used in the regular Python task.
             # The attributes include: image, command, args, resource, and env (env is unioned)
-            if container.name == self.pod_template.primary_container_name:
+            if container.name == cast(PodTemplate, self.pod_template).primary_container_name:
                 sdk_default_container = self._get_container(settings)
                 container.image = sdk_default_container.image
                 # clear existing commands
@@ -243,9 +243,9 @@ def _serialize_pod_spec(self, settings: SerializationSettings) -> Dict[str, Any]
                     container.env or []
                 )
             final_containers.append(container)
-        self.pod_template.pod_spec.containers = final_containers
+        cast(PodTemplate, self.pod_template).pod_spec.containers = final_containers
 
-        return ApiClient().sanitize_for_serialization(self.pod_template.pod_spec)
+        return ApiClient().sanitize_for_serialization(cast(PodTemplate, self.pod_template).pod_spec)
 
     def get_k8s_pod(self, settings: SerializationSettings) -> _task_model.K8sPod:
         if self.pod_template is None:

flytekit/core/type_engine.py

@@ -1152,7 +1152,7 @@ def to_python_value(self, ctx: FlyteContext, lv: Literal, expected_python_type:
 
     def guess_python_type(self, literal_type: LiteralType) -> type:
         if literal_type.union_type is not None:
-            return typing.Union[tuple(TypeEngine.guess_python_type(v) for v in literal_type.union_type.variants)]
+            return typing.Union[tuple(TypeEngine.guess_python_type(v) for v in literal_type.union_type.variants)]  # type: ignore
 
         raise ValueError(f"Union transformer cannot reverse {literal_type}")
 

flytekit/core/workflow.py

@@ -766,7 +766,7 @@ def wrapper(fn):
     if _workflow_function:
         return wrapper(_workflow_function)
     else:
-        return wrapper
+        return wrapper  # type: ignore
 
 
 class ReferenceWorkflow(ReferenceEntity, PythonFunctionWorkflow):  # type: ignore

tests/flytekit/unit/core/test_python_function_task.py

@@ -145,7 +145,7 @@ def test_pod_template():
         pod_template_name="A",
     )
     def func_with_pod_template(i: str):
-        print(i + 3)
+        print(i + "a")
 
     default_image = Image(name="default", fqn="docker.io/xyz", tag="some-git-hash")
     default_image_config = ImageConfig(default_image=default_image)