How to use _get_array_check_statistics method in pandera

Best Python code snippet using pandera_python

schema_statistics.py

Source:schema_statistics.py

...12    column_statistics = {13        col: {14            "dtype": dtype,15            "nullable": bool(nullable_columns[col]),16            "checks": _get_array_check_statistics(df[col], dtype),17        }18        for col, dtype in inferred_column_dtypes.items()19    }20    return {21        "columns": column_statistics if column_statistics else None,22        "index": infer_index_statistics(df.index),23    }24def infer_series_statistics(series: pd.Series) -> Dict[str, Any]:25    """Infer column and index statistics from a pandas Series."""26    dtype = _get_array_type(series)27    return {28        "dtype": dtype,29        "nullable": bool(series.isna().any()),30        "checks": _get_array_check_statistics(series, dtype),31        "name": series.name,32    }33def infer_index_statistics(index: Union[pd.Index, pd.MultiIndex]):34    """Infer index statistics given a pandas Index object."""35    def _index_stats(index_level):36        dtype = _get_array_type(index_level)37        return {38            "dtype": dtype,39            "nullable": bool(index_level.isna().any()),40            "checks": _get_array_check_statistics(index_level, dtype),41            "name": index_level.name,42        }43    if isinstance(index, pd.MultiIndex):44        index_statistics = [45            _index_stats(index.get_level_values(i))46            for i in range(index.nlevels)47        ]48    elif isinstance(index, pd.Index):49        index_statistics = [_index_stats(index)]50    else:51        warnings.warn(52            f"index type {type(index)} not recognized, skipping index inference",53            UserWarning,54        )55        index_statistics = []56    return index_statistics if index_statistics else None57def parse_check_statistics(check_stats: Union[Dict[str, Any], None]):58    """Convert check statistics to a list of Check objects."""59    if check_stats is None:60        return None61    checks = []62    for check_name, stats in check_stats.items():63        check = getattr(Check, check_name)64        try:65            checks.append(check(**stats))66        except TypeError:67            # if stats cannot be unpacked as key-word args, assume unary check.68            checks.append(check(stats))69    return checks if checks else None70def get_dataframe_schema_statistics(dataframe_schema):71    """Get statistical properties from dataframe schema."""72    statistics = {73        "columns": {74            col_name: {75                "dtype": column.dtype,76                "nullable": column.nullable,77                "coerce": column.coerce,78                "required": column.required,79                "regex": column.regex,80                "checks": parse_checks(column.checks),81                "unique": column.unique,82            }83            for col_name, column in dataframe_schema.columns.items()84        },85        "checks": parse_checks(dataframe_schema.checks),86        "index": (87            None88            if dataframe_schema.index is None89            else get_index_schema_statistics(dataframe_schema.index)90        ),91        "coerce": dataframe_schema.coerce,92    }93    return statistics94def _get_series_base_schema_statistics(series_schema_base):95    return {96        "dtype": series_schema_base.dtype,97        "nullable": series_schema_base.nullable,98        "checks": parse_checks(series_schema_base.checks),99        "coerce": series_schema_base.coerce,100        "name": series_schema_base.name,101    }102def get_index_schema_statistics(index_schema_component):103    """Get statistical properties of index schema component."""104    try:105        # get index components from MultiIndex106        index_components = index_schema_component.indexes107    except AttributeError:108        index_components = [index_schema_component]109    return [110        _get_series_base_schema_statistics(index_component)111        for index_component in index_components112    ]113def get_series_schema_statistics(series_schema):114    """Get statistical properties from series schema."""115    return _get_series_base_schema_statistics(series_schema)116def parse_checks(checks) -> Union[Dict[str, Any], None]:117    """Convert Check object to check statistics."""118    check_statistics = {}119    _check_memo = {}120    for check in checks:121        if check not in Check:122            warnings.warn(123                "Only registered checks may be serialized to statistics. "124                "Did you forget to register it with the extension API? "125                f"Check `{check.name}` will be skipped."126            )127            continue128        check_statistics[check.name] = (129            {} if check.statistics is None else check.statistics130        )131        _check_memo[check.name] = check132    # raise ValueError on incompatible checks133    if (134        "greater_than_or_equal_to" in check_statistics135        and "less_than_or_equal_to" in check_statistics136    ):137        min_value = check_statistics.get(138            "greater_than_or_equal_to", float("-inf")139        )["min_value"]140        max_value = check_statistics.get(141            "less_than_or_equal_to", float("inf")142        )["max_value"]143        if min_value > max_value:144            raise ValueError(145                f"checks {_check_memo['greater_than_or_equal_to']} "146                f"and {_check_memo['less_than_or_equal_to']} are incompatible, reason: "147                f"min value {min_value} > max value {max_value}"148            )149    return check_statistics if check_statistics else None150def _get_array_type(x):151    # get most granular type possible152    data_type = pandas_engine.Engine.dtype(x.dtype)153    # for object arrays, try to infer dtype154    if data_type is pandas_engine.Engine.dtype("object"):155        inferred_alias = pd.api.types.infer_dtype(x, skipna=True)156        if inferred_alias != "string":157            data_type = pandas_engine.Engine.dtype(inferred_alias)158    return data_type159def _get_array_check_statistics(160    x, data_type: dtypes.DataType161) -> Union[Dict[str, Any], None]:162    """Get check statistics from an array-like object."""163    if dtypes.is_datetime(data_type):164        check_stats = {165            "greater_than_or_equal_to": x.min(),166            "less_than_or_equal_to": x.max(),167        }168    elif dtypes.is_numeric(data_type) and not dtypes.is_bool(data_type):169        check_stats = {170            "greater_than_or_equal_to": float(x.min()),171            "less_than_or_equal_to": float(x.max()),172        }173    elif dtypes.is_category(data_type):...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.