fix(pandas): apply parsers to Index fields and schema components

pandera/api/dataframe/model_components.py

@@ -78,6 +78,7 @@ def index_properties(
         self,
         dtype: Any,
         checks: CheckArg | None = None,
+        parsers: ParserArg | None = None,
         name: str | None = None,
     ) -> dict[str, Any]:
         """Create a schema_components.Index from a field."""
@@ -88,6 +89,7 @@ def index_properties(
             coerce=self.coerce,
             name=name,
             checks=checks,
+            parsers=parsers,
             title=self.title,
             description=self.description,
             default=self.default,

pandera/api/pandas/model.py

@@ -208,6 +208,7 @@ def _build_columns_index(
                     field.index_properties(
                         dtype,
                         checks=field_checks,
+                        parsers=field_parsers,
                         name=field_name,
                     )
                     if field

pandera/backends/pandas/components.py

@@ -318,7 +318,9 @@ def validate(
 
         error_handler = ErrorHandler(lazy)
 
-        if schema.coerce:
+        # if the index has parsers, defer coercion to the array backend,
+        # which runs parsers before coercing
+        if schema.coerce and not schema.parsers:
             try:
                 check_obj.index = schema.coerce_dtype(check_obj.index)
             except SchemaError as exc:
@@ -340,6 +342,8 @@ def validate(
                 inplace=inplace,
             )
             assert is_field(_validated_obj)
+            if schema.parsers:
+                check_obj.index = pd.Index(_validated_obj)
         except SchemaError as exc:
             error_handler.collect_error(
                 get_error_category(exc.reason_code),

pandera/backends/pandas/container.py

@@ -786,7 +786,13 @@ def _try_coercion(coerce_fn, obj):
 
         if schema.dtype is not None:
             obj = _try_coercion(_coerce_df_dtype, obj)
-        if schema.index is not None and (schema.index.coerce or schema.coerce):
+        if (
+            schema.index is not None
+            and (schema.index.coerce or schema.coerce)
+            # coercion of an index with parsers is deferred to index-level
+            # validation so that parsers run before coercion
+            and not getattr(schema.index, "parsers", None)
+        ):
             index_schema = copy.deepcopy(schema.index)
             if schema.coerce:
                 # coercing at the dataframe-level should apply index coercion

tests/pandas/test_parsers.py

@@ -12,7 +12,7 @@
 from pandera.api.pandas.container import DataFrameSchema
 from pandera.api.parsers import Parser
 from pandera.engines.pandas_engine import PANDAS_3_0_0_PLUS
-from pandera.typing import Series
+from pandera.typing import Index, Series
 
 
 def test_dataframe_schema_parse() -> None:
@@ -310,3 +310,59 @@ def test_column_parser_with_inferred_schema_coercion():
         [-13123.0, -12.0, np.nan], name="col1", dtype="float64"
     )
     pd.testing.assert_series_equal(validated["col1"], expected)
+
+
+def test_parser_on_dataframe_model_index_field():
+    """``@pa.parser`` should be applied to fields annotated as Index
+    (issue #1684)."""
+
+    class Model(pa.DataFrameModel):
+        idx: Index[int]
+        col: Series[int]
+
+        class Config:
+            coerce = True
+
+        @pa.parser("idx")
+        @classmethod
+        def double(cls, series: pd.Series) -> pd.Series:
+            return series * 2
+
+        @pa.parser("col")
+        @classmethod
+        def triple(cls, series: pd.Series) -> pd.Series:
+            return series * 3
+
+    df = pd.DataFrame({"col": [1, 2, 3]}, index=pd.Index([1, 2, 3]))
+    validated = Model.validate(df)
+    assert validated.index.tolist() == [2, 4, 6]
+    assert validated["col"].tolist() == [3, 6, 9]
+
+
+def test_index_parser():
+    """Parsers on an Index schema component should transform the index."""
+    schema = DataFrameSchema(
+        columns={"col": pa.Column(int)},
+        index=pa.Index(int, parsers=Parser(lambda s: s * 2), name="idx"),
+    )
+    df = pd.DataFrame(
+        {"col": [1, 2, 3]}, index=pd.Index([1, 2, 3], name="idx")
+    )
+    validated = schema.validate(df)
+    assert validated.index.tolist() == [2, 4, 6]
+    assert validated.index.name == "idx"
+
+
+def test_index_parser_output_needs_coercion():
+    """Index parsers should run before dtype coercion."""
+    schema = DataFrameSchema(
+        columns={"col": pa.Column(int)},
+        index=pa.Index(
+            float,
+            parsers=Parser(lambda s: s.str.replace(",", ".")),
+            coerce=True,
+        ),
+    )
+    df = pd.DataFrame({"col": [1, 2]}, index=pd.Index(["1,5", "2,5"]))
+    validated = schema.validate(df)
+    assert validated.index.tolist() == [1.5, 2.5]