How to use _get_dtype_kwargs method in pandera

Best Python code snippet using pandera_python

model.py

Source:model.py Github

copy

Full Screen

...241 "Cannot specify redundant 'dtype_kwargs' "242 + f"for {annotation.raw_annotation}."243 + "\n Usage Tip: Drop 'typing.Annotated'."244 )245 dtype_kwargs = _get_dtype_kwargs(annotation)246 dtype = annotation.arg(**dtype_kwargs) # type: ignore247 elif annotation.default_dtype:248 dtype = annotation.default_dtype249 else:250 dtype = annotation.arg251 dtype = None if dtype is Any else dtype252 if (253 annotation.origin in SERIES_TYPES254 or annotation.raw_annotation in SERIES_TYPES255 ):256 col_constructor = (257 field.to_column if field else schema_components.Column258 )259 if check_name is False:260 raise SchemaInitError(261 f"'check_name' is not supported for {field_name}."262 )263 columns[field_name] = col_constructor( # type: ignore264 dtype,265 required=not annotation.optional,266 checks=field_checks,267 name=field_name,268 )269 elif (270 annotation.origin in INDEX_TYPES271 or annotation.raw_annotation in INDEX_TYPES272 ):273 if annotation.optional:274 raise SchemaInitError(275 f"Index '{field_name}' cannot be Optional."276 )277 if check_name is False or (278 # default single index279 check_name is None280 and index_count == 1281 ):282 field_name = None # type:ignore283 index_constructor = (284 field.to_index if field else schema_components.Index285 )286 index = index_constructor( # type: ignore287 dtype, checks=field_checks, name=field_name288 )289 indices.append(index)290 else: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)438 matched.update(filter(pattern.match, seq))439 return matched440def _get_dtype_kwargs(annotation: AnnotationInfo) -> Dict[str, Any]:441 sig = inspect.signature(annotation.arg) # type: ignore442 dtype_arg_names = list(sig.parameters.keys())443 if len(annotation.metadata) != len(dtype_arg_names): # type: ignore444 raise TypeError(445 f"Annotation '{annotation.arg.__name__}' requires " # type: ignore446 + f"all positional arguments {dtype_arg_names}."447 )...

Full Screen

Full Screen

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.

LambdaTest Learning Hubs:

YouTube

You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.

Run pandera automation tests on LambdaTest cloud grid

Perform automation testing on 3000+ real desktop and mobile devices online.

Try LambdaTest Now !!

Get 100 minutes of automation test minutes FREE!!

Next-Gen App & Browser Testing Cloud

Was this article helpful?

Helpful

NotHelpful