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

How to use _build_schema_index method in pandera

Best Python code snippet using pandera_python

model.py

Source:model.py

...291                raise SchemaInitError(292                    f"Invalid annotation '{field_name}: "293                    f"{annotation.raw_annotation}'"294                )295        return columns, _build_schema_index(indices, **multiindex_kwargs)296    @classmethod297    def _get_model_attrs(cls) -> Dict[str, Any]:298        """Return all attributes.299        Similar to inspect.get_members but bypass descriptors __get__.300        """301        bases = inspect.getmro(cls)[:-1]  # bases -> SchemaModel -> object302        attrs = {}303        for base in reversed(bases):304            attrs.update(base.__dict__)305        return attrs306    @classmethod307    def _collect_fields(cls) -> Dict[str, Tuple[AnnotationInfo, FieldInfo]]:308        """Centralize publicly named fields and their corresponding annotations."""309        annotations = get_type_hints(  # pylint:disable=unexpected-keyword-arg310            cls, include_extras=True311        )312        attrs = cls._get_model_attrs()313        missing = []314        for name, attr in attrs.items():315            if inspect.isroutine(attr):316                continue317            if not _is_field(name):318                annotations.pop(name, None)319            elif name not in annotations:320                missing.append(name)321        if missing:322            raise SchemaInitError(f"Found missing annotations: {missing}")323        fields = {}324        for field_name, annotation in annotations.items():325            field = attrs[field_name]  # __init_subclass__ guarantees existence326            if not isinstance(field, FieldInfo):327                raise SchemaInitError(328                    f"'{field_name}' can only be assigned a 'Field', "329                    + f"not a '{type(field)}.'"330                )331            fields[field.name] = (AnnotationInfo(annotation), field)332        return fields333    @classmethod334    def _collect_config_and_extras(335        cls,336    ) -> Tuple[Type[BaseConfig], Dict[str, Any]]:337        """Collect config options from bases, splitting off unknown options."""338        bases = inspect.getmro(cls)[:-1]339        bases = typing.cast(Tuple[Type[SchemaModel]], bases)340        root_model, *models = reversed(bases)341        options, extras = _extract_config_options_and_extras(root_model.Config)342        for model in models:343            config = getattr(model, _CONFIG_KEY, {})344            base_options, base_extras = _extract_config_options_and_extras(345                config346            )347            options.update(base_options)348            extras.update(base_extras)349        return type("Config", (BaseConfig,), options), extras350    @classmethod351    def _collect_check_infos(cls, key: str) -> List[CheckInfo]:352        """Collect inherited check metadata from bases.353        Inherited classmethods are not in cls.__dict__, that's why we need to354        walk the inheritance tree.355        """356        bases = inspect.getmro(cls)[:-2]  # bases -> SchemaModel -> object357        bases = typing.cast(Tuple[Type[SchemaModel]], bases)358        method_names = set()359        check_infos = []360        for base in bases:361            for attr_name, attr_value in vars(base).items():362                check_info = getattr(attr_value, key, None)363                if not isinstance(check_info, CheckInfo):364                    continue365                if attr_name in method_names:  # check overridden by subclass366                    continue367                method_names.add(attr_name)368                check_infos.append(check_info)369        return check_infos370    @classmethod371    def _extract_checks(372        cls, check_infos: List[FieldCheckInfo], field_names: List[str]373    ) -> Dict[str, List[Check]]:374        """Collect field annotations from bases in mro reverse order."""375        checks: Dict[str, List[Check]] = {}376        for check_info in check_infos:377            check_info_fields = {378                field.name if isinstance(field, FieldInfo) else field379                for field in check_info.fields380            }381            if check_info.regex:382                matched = _regex_filter(field_names, check_info_fields)383            else:384                matched = check_info_fields385            check_ = check_info.to_check(cls)386            for field in matched:387                if field not in field_names:388                    raise SchemaInitError(389                        f"Check {check_.name} is assigned to a non-existing field '{field}'."390                    )391                if field not in checks:392                    checks[field] = []393                checks[field].append(check_)394        return checks395    @classmethod396    def _extract_df_checks(cls, check_infos: List[CheckInfo]) -> List[Check]:397        """Collect field annotations from bases in mro reverse order."""398        return [check_info.to_check(cls) for check_info in check_infos]399    @classmethod400    def __get_validators__(cls):401        yield cls._pydantic_validate402    @classmethod403    def _pydantic_validate(cls, schema_model: Any) -> "SchemaModel":404        """Verify that the input is a compatible schema model."""405        if not inspect.isclass(schema_model):  # type: ignore406            raise TypeError(f"{schema_model} is not a pandera.SchemaModel")407        if not issubclass(schema_model, cls):  # type: ignore408            raise TypeError(f"{schema_model} does not inherit {cls}.")409        try:410            schema_model.to_schema()411        except SchemaInitError as exc:412            raise ValueError(413                f"Cannot use {cls} as a pydantic type as its "414                "SchemaModel cannot be converted to a DataFrameSchema.\n"415                f"Please revisit the model to address the following errors:"416                f"\n{exc}"417            ) from exc418        return cast("SchemaModel", schema_model)419    @classmethod420    def __modify_schema__(cls, field_schema):421        """Update pydantic field schema."""422        field_schema.update(to_json_schema(cls.to_schema()))423def _build_schema_index(424    indices: List[schema_components.Index], **multiindex_kwargs: Any425) -> Optional[SchemaIndex]:426    index: Optional[SchemaIndex] = None427    if indices:428        if len(indices) == 1:429            index = indices[0]430        else:431            index = schema_components.MultiIndex(indices, **multiindex_kwargs)432    return index433def _regex_filter(seq: Iterable, regexps: Iterable[str]) -> Set[str]:434    """Filter items matching at least one of the regexes."""435    matched: Set[str] = set()436    for regex in regexps:437        pattern = re.compile(regex)...

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.