How to use handle_stat_dtype method in pandera

Best Python code snippet using pandera_python

io.py

Source:io.py

...27def _get_qualified_name(cls: type) -> str:28    return f"{cls.__module__}.{cls.__qualname__}"29def _serialize_check_stats(check_stats, dtype=None):30    """Serialize check statistics into json/yaml-compatible format."""31    def handle_stat_dtype(stat):32        if pandas_engine.Engine.dtype(dtypes.DateTime).check(33            dtype34        ) and hasattr(stat, "strftime"):35            # try serializing stat as a string if it's datetime-like,36            # otherwise return original value37            return stat.strftime(DATETIME_FORMAT)38        elif pandas_engine.Engine.dtype(dtypes.Timedelta).check(39            dtype40        ) and hasattr(stat, "delta"):41            # try serializing stat into an int in nanoseconds if it's42            # timedelta-like, otherwise return original value43            return stat.delta44        return stat45    # for unary checks, return a single value instead of a dictionary46    if len(check_stats) == 1:47        return handle_stat_dtype(list(check_stats.values())[0])48    # otherwise return a dictionary of keyword args needed to create the Check49    serialized_check_stats = {}50    for arg, stat in check_stats.items():51        serialized_check_stats[arg] = handle_stat_dtype(stat)52    return serialized_check_stats53def _serialize_dataframe_stats(dataframe_checks):54    """55    Serialize global dataframe check statistics into json/yaml-compatible56    format.57    """58    serialized_checks = {}59    for check_name, check_stats in dataframe_checks.items():60        # The case that `check_name` is not registered is handled in61        # `parse_checks` so we know that `check_name` exists.62        # infer dtype of statistics and serialize them63        serialized_checks[check_name] = _serialize_check_stats(check_stats)64    return serialized_checks65def _serialize_component_stats(component_stats):66    """67    Serialize column or index statistics into json/yaml-compatible format.68    """69    serialized_checks = None70    if component_stats["checks"] is not None:71        serialized_checks = {}72        for check_name, check_stats in component_stats["checks"].items():73            serialized_checks[check_name] = _serialize_check_stats(74                check_stats, component_stats["dtype"]75            )76    dtype = component_stats.get("dtype")77    if dtype:78        dtype = str(dtype)79    return {80        "dtype": dtype,81        "nullable": component_stats["nullable"],82        "checks": serialized_checks,83        **{84            key: component_stats.get(key)85            for key in [86                "name",87                "unique",88                "coerce",89                "required",90                "regex",91            ]92            if key in component_stats93        },94    }95def _serialize_schema(dataframe_schema):96    """Serialize dataframe schema into into json/yaml-compatible format."""97    from pandera import __version__  # pylint: disable=import-outside-toplevel98    statistics = get_dataframe_schema_statistics(dataframe_schema)99    columns, index, checks = None, None, None100    if statistics["columns"] is not None:101        columns = {102            col_name: _serialize_component_stats(column_stats)103            for col_name, column_stats in statistics["columns"].items()104        }105    if statistics["index"] is not None:106        index = [107            _serialize_component_stats(index_stats)108            for index_stats in statistics["index"]109        ]110    if statistics["checks"] is not None:111        checks = _serialize_dataframe_stats(statistics["checks"])112    return {113        "schema_type": "dataframe",114        "version": __version__,115        "columns": columns,116        "checks": checks,117        "index": index,118        "coerce": dataframe_schema.coerce,119        "strict": dataframe_schema.strict,120        "unique": dataframe_schema.unique,121    }122def _deserialize_check_stats(check, serialized_check_stats, dtype=None):123    def handle_stat_dtype(stat):124        try:125            if pandas_engine.Engine.dtype(dtypes.DateTime).check(dtype):126                return pd.to_datetime(stat, format=DATETIME_FORMAT)127            elif pandas_engine.Engine.dtype(dtypes.Timedelta).check(dtype):128                # serialize to int in nanoseconds129                return pd.to_timedelta(stat, unit="ns")130        except (TypeError, ValueError):131            return stat132        return stat133    if isinstance(serialized_check_stats, dict):134        # handle case where serialized check stats are in the form of a135        # dictionary mapping Check arg names to values.136        check_stats = {}137        for arg, stat in serialized_check_stats.items():138            check_stats[arg] = handle_stat_dtype(stat)139        return check(**check_stats)140    # otherwise assume unary check function signature141    return check(handle_stat_dtype(serialized_check_stats))142def _deserialize_component_stats(serialized_component_stats):143    dtype = serialized_component_stats.get("dtype")144    if dtype:145        dtype = pandas_engine.Engine.dtype(dtype)146    checks = serialized_component_stats.get("checks")147    if checks is not None:148        checks = [149            _deserialize_check_stats(150                getattr(Check, check_name), check_stats, dtype151            )152            for check_name, check_stats in checks.items()153        ]154    return {155        "dtype": dtype,...

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.