Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use _handle_schema_error method in pandera

Best Python code snippet using pandera_python

decorators.py

Source:decorators.py

...63    ):64        # Don't include "self" / "cls" argument65        arg_spec_args = arg_spec_args[1:]66    return arg_spec_args67def _handle_schema_error(68    decorator_name,69    fn: Callable,70    schema: Union[schemas.DataFrameSchema, schemas.SeriesSchema],71    arg_df: pd.DataFrame,72    schema_error: errors.SchemaError,73) -> NoReturn:74    """Reraise schema validation error with decorator context.75    :param fn: check the DataFrame or Series input of this function.76    :param schema: dataframe/series schema object77    :param arg_df: dataframe/series we are validating.78    :param schema_error: original exception.79    :raises SchemaError: when ``DataFrame`` violates built-in or custom80        checks.81    """82    msg = f"error in {decorator_name} decorator of function '{fn.__name__}': {schema_error}"83    raise errors.SchemaError(84        schema,85        arg_df,86        msg,87        failure_cases=schema_error.failure_cases,88        check=schema_error.check,89        check_index=schema_error.check_index,90    ) from schema_error91def check_input(92    schema: Schemas,93    obj_getter: Optional[InputGetter] = None,94    head: Optional[int] = None,95    tail: Optional[int] = None,96    sample: Optional[int] = None,97    random_state: Optional[int] = None,98    lazy: bool = False,99    inplace: bool = False,100) -> Callable[[F], F]:101    # pylint: disable=duplicate-code102    """Validate function argument when function is called.103    This is a decorator function that validates the schema of a dataframe104    argument in a function.105    :param schema: dataframe/series schema object106    :param obj_getter:  (Default value = None) if int, obj_getter refers to the107        the index of the pandas dataframe/series to be validated in the args108        part of the function signature. If str, obj_getter refers to the109        argument name of the pandas dataframe/series in the function signature.110        This works even if the series/dataframe is passed in as a positional111        argument when the function is called. If None, assumes that the112        dataframe/series is the first argument of the decorated function113    :param head: validate the first n rows. Rows overlapping with `tail` or114        `sample` are de-duplicated.115    :param tail: validate the last n rows. Rows overlapping with `head` or116        `sample` are de-duplicated.117    :param sample: validate a random sample of n rows. Rows overlapping118        with `head` or `tail` are de-duplicated.119    :param random_state: random seed for the ``sample`` argument.120    :param lazy: if True, lazily evaluates dataframe against all validation121        checks and raises a ``SchemaErrors``. Otherwise, raise122        ``SchemaError`` as soon as one occurs.123    :param inplace: if True, applies coercion to the object of validation,124        otherwise creates a copy of the data.125    :returns: wrapped function126    :example:127    Check the input of a decorated function.128    >>> import pandas as pd129    >>> import pandera as pa130    >>>131    >>>132    >>> schema = pa.DataFrameSchema({"column": pa.Column(int)})133    >>>134    >>> @pa.check_input(schema)135    ... def transform_data(df: pd.DataFrame) -> pd.DataFrame:136    ...     df["doubled_column"] = df["column"] * 2137    ...     return df138    >>>139    >>> df = pd.DataFrame({140    ...     "column": range(5),141    ... })142    >>>143    >>> transform_data(df)144       column  doubled_column145    0       0               0146    1       1               2147    2       2               4148    3       3               6149    4       4               8150    See :ref:`here<decorators>` for more usage details.151    """152    @wrapt.decorator153    def _wrapper(154        fn: Callable,155        instance: Union[None, Any],156        args: Tuple[Any, ...],157        kwargs: Dict[str, Any],158    ):159        # pylint: disable=unused-argument160        """Check pandas DataFrame or Series before calling the function.161        :param fn: check the DataFrame or Series input of this function162        :param instance: the object to which the wrapped function was bound163            when it was called. Only applies to methods.164        :param args: the list of positional arguments supplied when the165            decorated function was called.166        :param kwargs: the dictionary of keyword arguments supplied when the167            decorated function was called.168        """169        args = list(args)170        validate_args = (head, tail, sample, random_state, lazy, inplace)171        if isinstance(obj_getter, int):172            try:173                args[obj_getter] = schema.validate(args[obj_getter])174            except IndexError as exc:175                raise IndexError(176                    f"error in check_input decorator of function '{fn.__name__}': the "177                    f"index '{obj_getter}' was supplied to the check but this "178                    f"function accepts '{len(_get_fn_argnames(fn))}' arguments, so the maximum "179                    f"index is 'max(0, len(_get_fn_argnames(fn)) - 1)'. The full error is: '{exc}'"180                ) from exc181        elif isinstance(obj_getter, str):182            if obj_getter in kwargs:183                kwargs[obj_getter] = schema.validate(184                    kwargs[obj_getter], *validate_args185                )186            else:187                arg_spec_args = _get_fn_argnames(fn)188                args_dict = OrderedDict(zip(arg_spec_args, args))189                args_dict[obj_getter] = schema.validate(190                    args_dict[obj_getter], *validate_args191                )192                args = list(args_dict.values())193        elif obj_getter is None and args:194            try:195                args[0] = schema.validate(args[0], *validate_args)196            except errors.SchemaError as e:197                _handle_schema_error("check_input", fn, schema, args[0], e)198        elif obj_getter is None and kwargs:199            # get the first key in the same order specified in the200            # function argument.201            args_names = _get_fn_argnames(fn)202            try:203                kwargs[args_names[0]] = schema.validate(204                    kwargs[args_names[0]], *validate_args205                )206            except errors.SchemaError as e:207                _handle_schema_error(208                    "check_input", fn, schema, kwargs[args_names[0]], e209                )210        else:211            raise TypeError(212                f"obj_getter is unrecognized type: {type(obj_getter)}"213            )214        return fn(*args, **kwargs)215    return _wrapper216def check_output(217    schema: Schemas,218    obj_getter: Optional[OutputGetter] = None,219    head: Optional[int] = None,220    tail: Optional[int] = None,221    sample: Optional[int] = None,222    random_state: Optional[int] = None,223    lazy: bool = False,224    inplace: bool = False,225) -> Callable[[F], F]:226    # pylint: disable=duplicate-code227    """Validate function output.228    Similar to input validator, but validates the output of the decorated229    function.230    :param schema: dataframe/series schema object231    :param obj_getter:  (Default value = None) if int, assumes that the output232        of the decorated function is a list-like object, where obj_getter is233        the index of the pandas data dataframe/series to be validated. If str,234        expects that the output is a dict-like object, and obj_getter is the235        key pointing to the dataframe/series to be validated. If a callable is236        supplied, it expects the output of decorated function and should return237        the dataframe/series to be validated.238    :param head: validate the first n rows. Rows overlapping with `tail` or239        `sample` are de-duplicated.240    :param tail: validate the last n rows. Rows overlapping with `head` or241        `sample` are de-duplicated.242    :param sample: validate a random sample of n rows. Rows overlapping243        with `head` or `tail` are de-duplicated.244    :param random_state: random seed for the ``sample`` argument.245    :param lazy: if True, lazily evaluates dataframe against all validation246        checks and raises a ``SchemaErrors``. Otherwise, raise247        ``SchemaError`` as soon as one occurs.248    :param inplace: if True, applies coercion to the object of validation,249            otherwise creates a copy of the data.250    :returns: wrapped function251    :example:252    Check the output a decorated function.253    >>> import pandas as pd254    >>> import pandera as pa255    >>>256    >>>257    >>> schema = pa.DataFrameSchema(258    ...     columns={"doubled_column": pa.Column(int)},259    ...     checks=pa.Check(260    ...         lambda df: df["doubled_column"] == df["column"] * 2261    ...     )262    ... )263    >>>264    >>> @pa.check_output(schema)265    ... def transform_data(df: pd.DataFrame) -> pd.DataFrame:266    ...     df["doubled_column"] = df["column"] * 2267    ...     return df268    >>>269    >>> df = pd.DataFrame({"column": range(5)})270    >>>271    >>> transform_data(df)272       column  doubled_column273    0       0               0274    1       1               2275    2       2               4276    3       3               6277    4       4               8278    See :ref:`here<decorators>` for more usage details.279    """280    def validate(out: Any, fn: Callable) -> None:281        if obj_getter is None:282            obj = out283        elif isinstance(obj_getter, (int, str)):284            obj = out[obj_getter]285        elif callable(obj_getter):286            obj = obj_getter(out)287        else:288            raise TypeError(289                f"obj_getter is unrecognized type: {type(obj_getter)}"290            )291        try:292            schema.validate(293                obj, head, tail, sample, random_state, lazy, inplace294            )295        except errors.SchemaError as e:296            _handle_schema_error("check_output", fn, schema, obj, e)297    @wrapt.decorator298    def _wrapper(299        fn: Callable,300        instance: Union[None, Any],301        args: Tuple[Any, ...],302        kwargs: Dict[str, Any],303    ):304        # pylint: disable=unused-argument305        """Check pandas DataFrame or Series before calling the function.306        :param fn: check the DataFrame or Series output of this function307        :param instance: the object to which the wrapped function was bound308            when it was called. Only applies to methods.309        :param args: the list of positional arguments supplied when the310            decorated function was called.311        :param kwargs: the dictionary of keyword arguments supplied when the312            decorated function was called.313        """314        if inspect.iscoroutinefunction(fn):315            async def aio_wrapper():316                res = await fn(*args, **kwargs)317                validate(res, fn)318                return res319            return aio_wrapper()320        else:321            out = fn(*args, **kwargs)322            validate(out, fn)323            return out324    return _wrapper325def check_io(326    head: int = None,327    tail: int = None,328    sample: int = None,329    random_state: int = None,330    lazy: bool = False,331    inplace: bool = False,332    out: Union[333        Schemas,334        Tuple[OutputGetter, Schemas],335        List[Tuple[OutputGetter, Schemas]],336    ] = None,337    **inputs: Schemas,338) -> Callable[[F], F]:339    """Check schema for multiple inputs and outputs.340    See :ref:`here<decorators>` for more usage details.341    :param head: validate the first n rows. Rows overlapping with `tail` or342        `sample` are de-duplicated.343    :param tail: validate the last n rows. Rows overlapping with `head` or344        `sample` are de-duplicated.345    :param sample: validate a random sample of n rows. Rows overlapping346        with `head` or `tail` are de-duplicated.347    :param random_state: random seed for the ``sample`` argument.348    :param lazy: if True, lazily evaluates dataframe against all validation349        checks and raises a ``SchemaErrors``. Otherwise, raise350        ``SchemaError`` as soon as one occurs.351    :param inplace: if True, applies coercion to the object of validation,352        otherwise creates a copy of the data.353    :param out: this should be a schema object if the function outputs a single354        dataframe/series. It can be a two-tuple, where the first element is355        a string, integer, or callable that fetches the pandas data structure356        in the output, and the second element is the schema to validate357        against. For multiple outputs, specify a list of two-tuples following358        the above structure.359    :param inputs: kwargs keys should be the argument name in the decorated360        function and values should be the schema used to validate the pandas361        data structure referenced by the argument name.362    :returns: wrapped function363    """364    check_args = (head, tail, sample, random_state, lazy, inplace)365    @wrapt.decorator366    def _wrapper(367        fn: Callable,368        instance: Union[None, Any],  # pylint: disable=unused-argument369        args: Tuple[Any, ...],370        kwargs: Dict[str, Any],371    ):372        """Check pandas DataFrame or Series before calling the function.373        :param fn: check the DataFrame or Series output of this function374        :param instance: the object to which the wrapped function was bound375            when it was called. Only applies to methods.376        :param args: the list of positional arguments supplied when the377            decorated function was called.378        :param kwargs: the dictionary of keyword arguments supplied when the379            decorated function was called.380        """381        out_schemas = out382        if isinstance(out, list):383            out_schemas = out384        elif isinstance(out, (schemas.DataFrameSchema, schemas.SeriesSchema)):385            out_schemas = [(None, out)]  # type: ignore386        elif isinstance(out, tuple):387            out_schemas = [out]388        elif out is None:389            out_schemas = []390        else:391            raise TypeError(392                f"type of out argument not recognized: {type(out)}"393            )394        wrapped_fn = fn395        for input_getter, input_schema in inputs.items():396            # pylint: disable=no-value-for-parameter397            wrapped_fn = check_input(398                input_schema, input_getter, *check_args  # type: ignore399            )(wrapped_fn)400        # pylint: disable=no-value-for-parameter401        for out_getter, out_schema in out_schemas:  # type: ignore402            wrapped_fn = check_output(out_schema, out_getter, *check_args)(403                wrapped_fn404            )405        return wrapped_fn(*args, **kwargs)406    return _wrapper407@overload408def check_types(409    wrapped: F,410    *,411    head: Optional[int] = None,412    tail: Optional[int] = None,413    sample: Optional[int] = None,414    random_state: Optional[int] = None,415    lazy: bool = False,416    inplace: bool = False,417) -> F:418    ...  # pragma: no cover419@overload420def check_types(421    wrapped: None = None,422    *,423    head: Optional[int] = None,424    tail: Optional[int] = None,425    sample: Optional[int] = None,426    random_state: Optional[int] = None,427    lazy: bool = False,428    inplace: bool = False,429) -> Callable[[F], F]:430    ...  # pragma: no cover431def check_types(432    wrapped=None,433    *,434    with_pydantic=False,435    head: Optional[int] = None,436    tail: Optional[int] = None,437    sample: Optional[int] = None,438    random_state: Optional[int] = None,439    lazy: bool = False,440    inplace: bool = False,441) -> Callable:442    """Validate function inputs and output based on type annotations.443    See the :ref:`User Guide <schema_models>` for more.444    :param wrapped: the function to decorate.445    :param with_pydantic: use ``pydantic.validate_arguments`` to validate446        inputs. This function is still needed to validate function outputs.447    :param head: validate the first n rows. Rows overlapping with `tail` or448        `sample` are de-duplicated.449    :param tail: validate the last n rows. Rows overlapping with `head` or450        `sample` are de-duplicated.451    :param sample: validate a random sample of n rows. Rows overlapping452        with `head` or `tail` are de-duplicated.453    :param random_state: random seed for the ``sample`` argument.454    :param lazy: if True, lazily evaluates dataframe against all validation455        checks and raises a ``SchemaErrors``. Otherwise, raise456        ``SchemaError`` as soon as one occurs.457    :param inplace: if True, applies coercion to the object of validation,458            otherwise creates a copy of the data.459    """460    # pylint: disable=too-many-locals461    if wrapped is None:462        return functools.partial(463            check_types,464            with_pydantic=with_pydantic,465            head=head,466            tail=tail,467            sample=sample,468            random_state=random_state,469            lazy=lazy,470            inplace=inplace,471        )472    # Front-load annotation parsing473    annotated_schema_models: Dict[str, Tuple[SchemaModel, AnnotationInfo]] = {}474    for arg_name_, annotation in typing.get_type_hints(wrapped).items():475        annotation_info = AnnotationInfo(annotation)476        if not annotation_info.is_generic_df:477            continue478        schema_model = cast(SchemaModel, annotation_info.arg)479        annotated_schema_models[arg_name_] = (schema_model, annotation_info)480    def _check_arg(arg_name: str, arg_value: Any) -> Any:481        """482        Validate function's argument if annoted with a schema, else483        pass-through.484        """485        schema_model, annotation_info = annotated_schema_models.get(486            arg_name, (None, None)487        )488        if schema_model is None:489            return arg_value490        if (491            annotation_info492            and not (annotation_info.optional and arg_value is None)493            # the pandera.schema attribute should only be available when494            # schema.validate has been called in the DF. There's probably495            # a better way of doing this496        ):497            config = schema_model.__config__498            data_container_type = annotation_info.origin499            schema = schema_model.to_schema()500            if data_container_type and config and config.from_format:501                arg_value = data_container_type.from_format(arg_value, config)502            if (503                arg_value.pandera.schema is None504                # don't re-validate a dataframe that contains the same exact505                # schema506                or arg_value.pandera.schema != schema507            ):508                try:509                    arg_value = schema.validate(510                        arg_value,511                        head,512                        tail,513                        sample,514                        random_state,515                        lazy,516                        inplace,517                    )518                except errors.SchemaError as e:519                    _handle_schema_error(520                        "check_types", wrapped, schema, arg_value, e521                    )522            if data_container_type and config and config.to_format:523                arg_value = data_container_type.to_format(arg_value, config)524            return arg_value525    sig = inspect.signature(wrapped)526    def validate_args(arguments: Dict[str, Any]) -> Dict[str, Any]:527        return {528            arg_name: _check_arg(arg_name, arg_value)529            for arg_name, arg_value in arguments.items()530        }531    def validate_inputs(532        instance: Optional[Any],533        args: Tuple[Any, ...],...

dispatcher.py

Source:dispatcher.py

...189                u'Unrecognized/expired response from peer: ' +190                six.text_type(msg))191    def _handle_nil_id_error_response(self, msg):192        self._log_error(msg)193    def _handle_schema_error(self, msg):194        msg_id = None195        if isinstance(msg.get('id'), (six.string_types, int)):196            msg_id = msg['id']197        self.rpc.socket_queue.put(198            self.rpc.definitions.error_response(199                msg_id, RpcErrors.invalid_request))200        self._log_error(u'Invalid Request: ' + six.text_type(msg))201    def _dispatch_batch(self, msgs):202        def _process():203            self._log_info(u'Received batch: ' + six.text_type(msgs))204            rfs = RpcForServices(self.rpc)205            promises = []206            nthreads = []207            for msg in msgs:...

common.py

Source:common.py

...20    @abc.abstractmethod21    async def _handle_message(self, msg):22        pass23    @abc.abstractmethod24    async def _handle_schema_error(self, msg, exc):25        pass26    async def process_one(self, sock):27        msg = await sock.recv_json()28        try:29            msg = self._request_schema.validate(msg)30        except schema.SchemaError as exc:31            reply = await self._handle_schema_error(32                msg, exc,33            )34        else:35            try:36                reply = await self._handle_message(msg)37            except Exception as exc:38                self.logger.error(39                    "handler failed for message %r",40                    msg,41                    exc_info=True,42                )43                reply = self._ise_response44        try:45            reply = self._response_schema.validate(reply)...

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.