Enhancement: drop invalid rows on validate with new param

docs/source/drop_invalid_rows.rst

@@ -0,0 +1,99 @@
+.. currentmodule:: pandera
+
+.. _drop_invalid_rows:
+
+Dropping Invalid Rows
+=====================
+
+*New in version 0.16.0*
+
+If you wish to use the validation step to remove invalid data, you can pass the
+``drop_invalid_rows=True`` argument to the ``schema`` object on creation. On ``schema.validate()``,
+if a data-level check fails, then that row which caused the failure will be removed from the dataframe
+when it is returned.
+
+``drop_invalid`` will prevent data-level schema errors being raised and will instead
+remove the rows which causes the failure.
+
+This functionality is available on ``DataFrameSchema``, ``SeriesSchema``, ``Column``,
+as well as ``DataFrameModel`` schemas.
+
+Dropping invalid rows with :class:`~pandera.api.pandas.container.DataFrameSchema`:
+
+.. testcode:: drop_invalid_rows_data_frame_schema
+
+   import pandas as pd
+   import pandera as pa
+
+   from pandera import Check, Column, DataFrameSchema
+
+   df = pd.DataFrame({"counter": ["1", "2", "3"]})
+   schema = DataFrameSchema(
+       {"counter": Column(int, checks=[Check(lambda x: x >= 3)])},
+       drop_invalid_rows=True,
+   )
+
+   schema.validate(df, lazy=True)
+
+Dropping invalid rows with :class:`~pandera.api.pandas.array.SeriesSchema`:
+
+.. testcode:: drop_invalid_rows_series_schema
+
+   import pandas as pd
+   import pandera as pa
+
+   from pandera import Check, SeriesSchema
+
+   series = pd.Series(["1", "2", "3"])
+   schema = SeriesSchema(
+       int,
+       checks=[Check(lambda x: x >= 3)],
+       drop_invalid_rows=True,
+   )
+
+   schema.validate(series, lazy=True)
+
+Dropping invalid rows with :class:`~pandera.api.pandas.components.Column`:
+
+.. testcode:: drop_invalid_rows_column
+
+   import pandas as pd
+   import pandera as pa
+
+   from pandera import Check, Column
+
+   df = pd.DataFrame({"counter": ["1", "2", "3"]})
+   schema = Column(
+       int,
+       name="counter",
+       drop_invalid_rows=True,
+       checks=[Check(lambda x: x >= 3)]
+   )
+
+   schema.validate(df, lazy=True)
+
+Dropping invalid rows with :class:`~pandera.api.pandas.model.DataFrameModel`:
+
+.. testcode:: drop_invalid_rows_data_frame_model
+
+    import pandas as pd
+    import pandera as pa
+
+    from pandera import Check, DataFrameModel, Field
+
+    class MySchema(DataFrameModel):
+        counter: int = Field(in_range={"min_value": 3, "max_value": 5})
+
+        class Config:
+            drop_invalid_rows = True
+
+
+    MySchema.validate(
+        pd.DataFrame({"counter": [1, 2, 3, 4, 5, 6]}), lazy=True
+    )
+
+.. note::
+    In order to use ``drop_invalid_rows=True``, ``lazy=True`` must
+    be passed to the ``schema.validate()``. :ref:`lazy_validation` enables all schema
+    errors to be collected and raised together, meaning all invalid rows can be dropped together.
+    This provides clear API for ensuring the validated dataframe contains only valid data.

docs/source/index.rst

@@ -360,6 +360,7 @@ page or reach out to the maintainers and pandera community on
    hypothesis
    dtypes
    decorators
+   drop_invalid_rows
    schema_inference
    lazy_validation
    data_synthesis_strategies

pandera/api/base/schema.py

@@ -32,6 +32,7 @@ def __init__(
         name=None,
         title=None,
         description=None,
+        drop_invalid_rows=False,
     ):
         """Abstract base schema initializer."""
         self.dtype = dtype
@@ -40,6 +41,7 @@ def __init__(
         self.name = name
         self.title = title
         self.description = description
+        self.drop_invalid_rows = drop_invalid_rows
 
     def validate(
         self,

pandera/api/pandas/array.py

@@ -37,6 +37,7 @@ def __init__(
         title: Optional[str] = None,
         description: Optional[str] = None,
         default: Optional[Any] = None,
+        drop_invalid_rows: bool = False,
     ) -> None:
         """Initialize array schema.
 
@@ -63,6 +64,8 @@ def __init__(
         :param title: A human-readable label for the series.
         :param description: An arbitrary textual description of the series.
         :param default: The default value for missing values in the series.
+        :param drop_invalid_rows: if True, drop invalid rows on validation.
+
         """
 
         super().__init__(
@@ -72,6 +75,7 @@ def __init__(
             name=name,
             title=title,
             description=description,
+            drop_invalid_rows=drop_invalid_rows,
         )
 
         if checks is None:
@@ -300,6 +304,7 @@ def __init__(
         title: Optional[str] = None,
         description: Optional[str] = None,
         default: Optional[Any] = None,
+        drop_invalid_rows: bool = False,
     ) -> None:
         """Initialize series schema base object.
 
@@ -327,6 +332,7 @@ def __init__(
         :param title: A human-readable label for the series.
         :param description: An arbitrary textual description of the series.
         :param default: The default value for missing values in the series.
+        :param drop_invalid_rows: if True, drop invalid rows on validation.
 
         """
         super().__init__(
@@ -340,6 +346,7 @@ def __init__(
             title,
             description,
             default,
+            drop_invalid_rows,
         )
         self.index = index
 

pandera/api/pandas/components.py

@@ -30,6 +30,7 @@ def __init__(
         title: Optional[str] = None,
         description: Optional[str] = None,
         default: Optional[Any] = None,
+        drop_invalid_rows: bool = False,
     ) -> None:
         """Create column validator object.
 
@@ -54,6 +55,7 @@ def __init__(
         :param title: A human-readable label for the column.
         :param description: An arbitrary textual description of the column.
         :param default: The default value for missing values in the column.
+        :param drop_invalid_rows: if True, drop invalid rows on validation.
 
         :raises SchemaInitError: if impossible to build schema from parameters
 
@@ -85,6 +87,7 @@ def __init__(
             title=title,
             description=description,
             default=default,
+            drop_invalid_rows=drop_invalid_rows,
         )
         if (
             name is not None

pandera/api/pandas/container.py

@@ -46,6 +46,7 @@ def __init__(
         unique_column_names: bool = False,
         title: Optional[str] = None,
         description: Optional[str] = None,
+        drop_invalid_rows: bool = False,
     ) -> None:
         """Initialize DataFrameSchema validator.
 
@@ -77,6 +78,7 @@ def __init__(
         :param unique_column_names: whether or not column names must be unique.
         :param title: A human-readable label for the schema.
         :param description: An arbitrary textual description of the schema.
+        :param drop_invalid_rows: if True, drop invalid rows on validation.
 
         :raises SchemaInitError: if impossible to build schema from parameters
 
@@ -152,6 +154,7 @@ def __init__(
         self._unique = unique
         self.report_duplicates = report_duplicates
         self.unique_column_names = unique_column_names
+        self.drop_invalid_rows = drop_invalid_rows
 
         # this attribute is not meant to be accessed by users and is explicitly
         # set to True in the case that a schema is created by infer_schema.

pandera/api/pandas/model.py

@@ -268,6 +268,7 @@ def to_schema(cls) -> DataFrameSchema:
                 "title": cls.__config__.title,
                 "description": cls.__config__.description or cls.__doc__,
                 "unique_column_names": cls.__config__.unique_column_names,
+                "drop_invalid_rows": cls.__config__.drop_invalid_rows,
             }
         cls.__schema__ = DataFrameSchema(
             columns,

pandera/api/pandas/model_config.py

@@ -21,6 +21,7 @@ class BaseConfig(BaseModelConfig):  # pylint:disable=R0903
     title: Optional[str] = None  #: human-readable label for schema
     description: Optional[str] = None  #: arbitrary textual description
     coerce: bool = False  #: coerce types of all schema components
+    drop_invalid_rows: bool = False  #: drop invalid rows on validation
 
     #: make sure certain column combinations are unique
     unique: Optional[Union[str, List[str]]] = None

pandera/backends/base/__init__.py

@@ -124,6 +124,10 @@ def failure_cases_metadata(
         """Get failure cases metadata for lazy validation."""
         raise NotImplementedError
 
+    def drop_invalid_rows(self, check_obj, error_handler):
+        """Remove invalid elements in a `check_obj` according to failures in caught by the `error_handler`"""
+        raise NotImplementedError
+
 
 class BaseCheckBackend(ABC):
     """Abstract base class for a check backend implementation."""

pandera/backends/pandas/array.py

@@ -20,6 +20,7 @@
     SchemaError,
     SchemaErrors,
     SchemaErrorReason,
+    SchemaDefinitionError,
 )
 
 
@@ -45,6 +46,11 @@ def validate(
         error_handler = SchemaErrorHandler(lazy)
         check_obj = self.preprocess(check_obj, inplace)
 
+        if getattr(schema, "drop_invalid_rows", False) and not lazy:
+            raise SchemaDefinitionError(
+                "When drop_invalid_rows is True, lazy must be set to True."
+            )
+
         # fill nans with `default` if it's present
         if hasattr(schema, "default") and pd.notna(schema.default):
             check_obj.fillna(schema.default, inplace=True)
@@ -55,6 +61,42 @@ def validate(
             except SchemaError as exc:
                 error_handler.collect_error(exc.reason_code, exc)
 
+        # run the core checks
+        error_handler = self.run_checks_and_handle_errors(
+            error_handler,
+            schema,
+            check_obj,
+            head,
+            tail,
+            sample,
+            random_state,
+        )
+
+        if lazy and error_handler.collected_errors:
+            if getattr(schema, "drop_invalid_rows", False):
+                check_obj = self.drop_invalid_rows(check_obj, error_handler)
+                return check_obj
+            else:
+                raise SchemaErrors(
+                    schema=schema,
+                    schema_errors=error_handler.collected_errors,
+                    data=check_obj,
+                )
+
+        return check_obj
+
+    def run_checks_and_handle_errors(
+        self,
+        error_handler,
+        schema,
+        check_obj,
+        head,
+        tail,
+        sample,
+        random_state,
+    ):
+        """Run checks on schema"""
+        # pylint: disable=too-many-locals
         field_obj_subsample = self.subsample(
             check_obj if is_field(check_obj) else check_obj[schema.name],
             head,
@@ -71,14 +113,15 @@ def validate(
             random_state,
         )
 
-        # run the core checks
-        for core_check, args in (
+        core_checks = [
             (self.check_name, (field_obj_subsample, schema)),
             (self.check_nullable, (field_obj_subsample, schema)),
             (self.check_unique, (field_obj_subsample, schema)),
             (self.check_dtype, (field_obj_subsample, schema)),
             (self.run_checks, (check_obj_subsample, schema)),
-        ):
+        ]
+
+        for core_check, args in core_checks:
             results = core_check(*args)
             if isinstance(results, CoreCheckResult):
                 results = [results]
@@ -106,13 +149,7 @@ def validate(
                         original_exc=result.original_exc,
                     )
 
-        if lazy and error_handler.collected_errors:
-            raise SchemaErrors(
-                schema=schema,
-                schema_errors=error_handler.collected_errors,
-                data=check_obj,
-            )
-        return check_obj
+        return error_handler
 
     def coerce_dtype(
         self,

pandera/backends/pandas/base.py

@@ -24,6 +24,7 @@
     scalar_failure_case,
 )
 from pandera.errors import FailureCaseMetadata, SchemaError, SchemaErrorReason
+from pandera.error_handlers import SchemaErrorHandler
 
 
 class ColumnInfo(NamedTuple):
@@ -149,3 +150,12 @@ def failure_cases_metadata(
             message=message,
             error_counts=error_counts,
         )
+
+    def drop_invalid_rows(self, check_obj, error_handler: SchemaErrorHandler):
+        """Remove invalid elements in a check obj according to failures in caught by the error handler."""
+        errors = error_handler.collected_errors
+        for err in errors:
+            check_obj = check_obj.loc[
+                ~check_obj.index.isin(err.failure_cases["index"])
+            ]
+        return check_obj