Fix plugin regressions

.github/workflows/pythonbuild.yml

@@ -72,7 +72,7 @@ jobs:
         python -m pip install --upgrade pip setuptools wheel
         make setup
         cd plugins/${{ matrix.plugin-names }}
-        pip install -r requirements.txt
+        pip install -e .
         pip freeze
     - name: Test with coverage
       run: |

flytekit/core/promise.py

@@ -770,7 +770,7 @@ def create_and_link_node(
             )
             used_inputs.add(k)
         except Exception as e:
-            raise AssertionError(f"Failed to Bind variable {k} for function {entity.name}.") from e
+            raise AssertionError(f"Failed to Bind variable {k} for function {entity.name}") from e
 
     extra_inputs = used_inputs ^ set(kwargs.keys())
     if len(extra_inputs) > 0:

plugins/flytekit-pandera/flytekitplugins/pandera/schema.py

@@ -11,6 +11,8 @@
 from flytekit.types.schema import FlyteSchema, PandasSchemaWriter, SchemaFormat, SchemaOpenMode
 from flytekit.types.schema.types import FlyteSchemaTransformer
 
+T = typing.TypeVar("T")
+
 
 class PanderaTransformer(TypeTransformer[pandera.typing.DataFrame]):
     _SUPPORTED_TYPES: typing.Dict[
@@ -53,6 +55,11 @@ def _get_schema_type(self, t: Type[pandera.typing.DataFrame]) -> SchemaType:
     def get_literal_type(self, t: Type[pandera.typing.DataFrame]) -> LiteralType:
         return LiteralType(schema=self._get_schema_type(t))
 
+    def assert_type(self, t: Type[T], v: T):
+        if not hasattr(t, "__origin__") and not isinstance(v, t):
+        # if not hasattr(t, "__origin__") and not isinstance(v, (t, pandas.DataFrame)):
+            raise TypeError(f"Type of Val '{v}' is not an instance of {t}")
+
     def to_literal(
         self,
         ctx: FlyteContext,

plugins/flytekit-pandera/tests/test_plugin.py

@@ -42,7 +42,7 @@ def my_wf() -> pandera.typing.DataFrame[OutSchema]:
     # raise error when defining workflow using invalid data
     invalid_df = pandas.DataFrame({"col1": [1, 2, 3], "col2": list("abc")})
 
-    with pytest.raises(pandera.errors.SchemaError):
+    with pytest.raises(AssertionError):
 
         @workflow
         def invalid_wf() -> pandera.typing.DataFrame[OutSchema]:
@@ -65,7 +65,7 @@ def transform2_noop(df: pandera.typing.DataFrame[IntermediateSchema]) -> pandera
     def wf_invalid_output(df: pandera.typing.DataFrame[InSchema]) -> pandera.typing.DataFrame[OutSchema]:
         return transform2_noop(df=transform1(df=df))
 
-    with pytest.raises(pandera.errors.SchemaError, match="column 'col4' not in dataframe"):
+    with pytest.raises(TypeError, match="^Failed to convert return value"):
         wf_invalid_output(df=valid_df)