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

How to use __modify_schema__ method in pandera

Best Python code snippet using pandera_python

validators.py

Source:validators.py

...24T = TypeVar("T")25class UniqueConstrainedList(ConstrainedList, List[T]):26    unique_items: Optional[bool] = None27    @classmethod28    def __modify_schema__(cls, field_schema: Dict[str, Any]) -> None:29        super().__modify_schema__(field_schema)30        update_not_none(field_schema, uniqueItems=cls.unique_items)31    @classmethod32    def __get_validators__(cls) -> Generator:  # noqa: B90233        yield from super().__get_validators__()34        if cls.unique_items is not None:35            yield cls.check_unique36    @classmethod37    def check_unique(cls, v: List[T]) -> List[T]:38        if len(set(v)) != len(v):39            raise ValueError("Items must be unique")40        return v41    def __init_subclass__(cls, **kwargs: Any) -> None:42        super().__init_subclass__(**kwargs)  # type:ignore43        # Copy generic argument (T) if not set explicitly44        # This makes a lot of assuptions about the internals of `typing`45        if "__orig_bases__" in cls.__dict__ and cls.__dict__["__orig_bases__"]:46            generic_base_cls = cls.__dict__["__orig_bases__"][0]47            if (not hasattr(cls, "item_type") or isinstance(cls.item_type, TypeVar)) and get_args(generic_base_cls):48                cls.item_type = get_args(generic_base_cls)[0]49        # Make sure __args__ is set50        assert hasattr(cls, "item_type"), "Missing a concrete value for generic type argument"51        cls.__args__ = (cls.item_type,)52    def __class_getitem__(cls, key: Any) -> Type:53        # Some magic to make sure that subclasses of this class still work as expected54        class Inst(cls):  # type: ignore55            item_type = key56            __args__ = (key,)57        return Inst58def unique_conlist(59    item_type: Type[T],60    *,61    min_items: Optional[int] = None,62    max_items: Optional[int] = None,63    unique_items: Optional[bool] = None,64) -> Type[List[T]]:65    namespace = {66        "min_items": min_items,67        "max_items": max_items,68        "unique_items": unique_items,69        "__origin__": list,  # Needed for pydantic to detect that this is a list70        "__args__": (item_type,),  # Needed for pydantic to detect the item type71    }72    # We use new_class to be able to deal with Generic types73    return new_class(74        "ConstrainedListValue",75        (UniqueConstrainedList[item_type],),76        {},77        lambda ns: ns.update(namespace),  # type:ignore78    )79def remove_empty_items(v: list) -> list:80    """Remove Falsy values from list.81    Sees dicts with all Falsy values as Falsy.82    This is used to allow people to submit list fields which are "empty" but are not really empty like:83    `[{}, None, {name:"", email:""}]`84    Example:85        >>> remove_empty_items([{}, None, [], {"a":""}])86        []87    """88    if v:89        return list(filter(lambda i: bool(i) and (not isinstance(i, dict) or any(i.values())), v))90    return v91class Accept:92    data: ClassVar[Optional[AcceptData]] = None93    class Values(strEnum):94        ACCEPTED = "ACCEPTED"95        INCOMPLETE = "INCOMPLETE"96    @classmethod97    def __modify_schema__(cls, field_schema: Dict[str, Any]) -> None:98        field_schema.update(99            format="accept",100            type="string",101            enum=[v.value for v in cls.Values],102            **({"data": cls.data} if cls.data else {}),103        )104    @classmethod105    def __get_validators__(cls) -> Generator:106        yield cls.enum_validator107        yield cls.must_be_complete108    @classmethod109    def enum_validator(cls, v: Any, field: "ModelField") -> "Accept":110        try:111            enum_v = cls.Values(v)112        except ValueError:113            # cls.Values should be an enum, so will be iterable114            raise EnumMemberError(enum_values=list(cls.Values))115        return enum_v.value116    @classmethod117    def must_be_complete(cls, v: str) -> bool:118        if v == "INCOMPLETE":119            raise ValueError("Not all tasks are done")120        return v == "ACCEPTED"121class Choice(strEnum):122    label: ClassVar[str]123    def __new__(cls, value: str, label: Optional[str] = None) -> "Choice":124        obj = str.__new__(cls, value)  # type:ignore125        obj._value_ = value126        obj.label = label or value127        return obj128    @classmethod129    def __modify_schema__(cls, field_schema: Dict[str, Any]) -> None:130        kwargs = {}131        options = dict(map(lambda i: (i.value, i.label), cls.__members__.values()))132        if not all(map(lambda o: o[0] == o[1], options.items())):133            kwargs["options"] = options134        field_schema.update(type="string", **kwargs)135class ChoiceList(UniqueConstrainedList[T]):136    @classmethod137    def __modify_schema__(cls, field_schema: Dict[str, Any]) -> None:138        super().__modify_schema__(field_schema)139        data: dict = {}140        cls.item_type.__modify_schema__(data)  # type: ignore141        field_schema.update(**{k: v for k, v in data.items() if k == "options"})142def choice_list(143    item_type: Type[Choice],144    *,145    min_items: Optional[int] = None,146    max_items: Optional[int] = None,147    unique_items: Optional[bool] = None,148) -> Type[List[T]]:149    namespace = {150        "min_items": min_items,151        "max_items": max_items,152        "unique_items": unique_items,153        "__origin__": list,  # Needed for pydantic to detect that this is a list154        "item_type": item_type,  # Needed for pydantic to detect the item type155        "__args__": (item_type,),  # Needed for pydantic to detect the item type156    }157    # We use new_class to be able to deal with Generic types158    return new_class(159        "ChoiceListValue",160        (ChoiceList[item_type],),161        {},162        lambda ns: ns.update(namespace),  # type:ignore163    )164class LongText(str):165    @classmethod166    def __modify_schema__(cls, field_schema: Dict[str, Any]) -> None:167        field_schema.update(format="long", type="string")168    @classmethod169    def __get_validators__(cls) -> Generator:170        yield str_validator171class Label(DisplayOnlyFieldType):172    @classmethod173    def __modify_schema__(cls, field_schema: Dict[str, Any]) -> None:174        field_schema.update(format="label", type="string")175class Summary(DisplayOnlyFieldType):176    data: ClassVar[Optional[SummaryData]] = None177    @classmethod178    def __modify_schema__(cls, field_schema: Dict[str, Any]) -> None:179        field_schema.update(format="summary", type="string", uniforms={"data": cls.data})180def summary(data: Optional[SummaryData] = None) -> Type[Summary]:181    namespace = {"data": data}...

fields.py

Source:fields.py

...12    @classmethod13    def __get_validators__(cls):14        yield cls.validate15    @classmethod16    def __modify_schema__(cls, field_schema):17        raise NotImplementedError()18    @classmethod19    def validate(cls, v):20        raise NotImplementedError()21    def __repr__(self):22        return f'{self.__class__.__name__}({super().__repr__()})'23class CXU(_Str):24    @classmethod25    def __modify_schema__(cls, field_schema):26        field_schema.update(pattern=cbu.cbu_rx.pattern,27                            example='2850590940090418135201')28    @classmethod29    def validate(cls, v):30        return cls(cbu.check(v))31CBU = CXU32CVU = CXU33class CXUAlias(_Str):34    @classmethod35    def __modify_schema__(cls, field_schema):36        field_schema.update(pattern=cxu_alias_rx.pattern,37                            example=['cbu-alias-example',38                                     'cbu.alias.example',39                                     'cbu.alias-example'])40    @classmethod41    def validate(cls, v):42        return check_cxu_alias(v) and cls(v)43CBUAlias = CXUAlias44CVUAlias = CXUAlias45class DNI(_Str):46    @classmethod47    def __modify_schema__(cls, field_schema):48        field_schema.update(pattern=dni.dni_rx.pattern,49                            example=['7.321.654', '7321654'])50    @classmethod51    def validate(cls, v):52        return cls(dni.check(v))53class CUIT(_Str):54    @classmethod55    def __modify_schema__(cls, field_schema):56        field_schema.update(pattern=cuit.cuit_rx.pattern,57                            example=['20-05536168-2', '20055361682'])58    @classmethod59    def validate(cls, v):60        return cls(cuit.check(v))61class ClientID(_Str):62    @classmethod63    def __modify_schema__(cls, field_schema):64        field_schema.update(pattern=client_id_rx.pattern,65                            example=['1234567', '123456789012'])66    @classmethod67    def validate(cls, v):68        return check_client_id(v) and cls(v)69class OriginID(_Str):70    @classmethod71    def __modify_schema__(cls, field_schema):72        field_schema.update(pattern=origin_id_rx.pattern,73                            example=['1234567', '123456789012345'])74    @classmethod75    def validate(cls, v):76        return check_origin_id(v) and cls(v)77class LogLevel(_Str):78    @classmethod79    def __modify_schema__(cls, field_schema):80        field_schema.update(pattern=r'^[a-zA-Z]$',81                            example=['DEBUG', 'INFO', 'WARNING'])82    @classmethod83    def validate(cls, v):...

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.