importwarningsfromabcimportABCMetafromcopyimportdeepcopyfromenumimportEnumfromfunctoolsimportpartialfrompathlibimportPathfromtypesimportFunctionType,prepare_class,resolve_basesfromtypingimport(TYPE_CHECKING,AbstractSet,Any,Callable,ClassVar,Dict,List,Mapping,Optional,Tuple,Type,TypeVar,Union,cast,no_type_check,overload,)fromtyping_extensionsimportdataclass_transformfrompydantic.v1.class_validatorsimportValidatorGroup,extract_root_validators,extract_validators,inherit_validatorsfrompydantic.v1.configimportBaseConfig,Extra,inherit_config,prepare_configfrompydantic.v1.error_wrappersimportErrorWrapper,ValidationErrorfrompydantic.v1.errorsimportConfigError,DictError,ExtraError,MissingErrorfrompydantic.v1.fieldsimport(MAPPING_LIKE_SHAPES,Field,ModelField,ModelPrivateAttr,PrivateAttr,Undefined,is_finalvar_with_default_val,)frompydantic.v1.jsonimportcustom_pydantic_encoder,pydantic_encoderfrompydantic.v1.parseimportProtocol,load_file,load_str_bytesfrompydantic.v1.schemaimportdefault_ref_template,model_schemafrompydantic.v1.typesimportPyObject,StrBytesfrompydantic.v1.typingimport(AnyCallable,get_args,get_origin,is_classvar,is_namedtuple,is_union,resolve_annotations,update_model_forward_refs,)frompydantic.v1.utilsimport(DUNDER_ATTRIBUTES,ROOT_KEY,ClassAttribute,GetterDict,Representation,ValueItems,generate_model_signature,is_valid_field,is_valid_private_name,lenient_issubclass,sequence_like,smart_deepcopy,unique_list,validate_field_name,)ifTYPE_CHECKING:frominspectimportSignaturefrompydantic.v1.class_validatorsimportValidatorListDictfrompydantic.v1.typesimportModelOrDcfrompydantic.v1.typingimport(AbstractSetIntStr,AnyClassMethod,CallableGenerator,DictAny,DictStrAny,MappingIntStrAny,ReprArgs,SetStr,TupleGenerator,)Model=TypeVar('Model',bound='BaseModel')__all__='BaseModel','create_model','validate_model'_T=TypeVar('_T')defvalidate_custom_root_type(fields:Dict[str,ModelField])->None:iflen(fields)>1:raiseValueError(f'{ROOT_KEY} cannot be mixed with other fields')defgenerate_hash_function(frozen:bool)->Optional[Callable[[Any],int]]:defhash_function(self_:Any)->int:returnhash(self_.__class__)+hash(tuple(self_.__dict__.values()))returnhash_functioniffrozenelseNone# If a field is of type `Callable`, its default value should be a function and cannot to ignored.ANNOTATED_FIELD_UNTOUCHED_TYPES:Tuple[Any,...]=(property,type,classmethod,staticmethod)# When creating a `BaseModel` instance, we bypass all the methods, properties... added to the modelUNTOUCHED_TYPES:Tuple[Any,...]=(FunctionType,)+ANNOTATED_FIELD_UNTOUCHED_TYPES# Note `ModelMetaclass` refers to `BaseModel`, but is also used to *create* `BaseModel`, so we need to add this extra# (somewhat hacky) boolean to keep track of whether we've created the `BaseModel` class yet, and therefore whether it's# safe to refer to it. If it *hasn't* been created, we assume that the `__new__` call we're in the middle of is for# the `BaseModel` class, since that's defined immediately after the metaclass._is_base_model_class_defined=False@dataclass_transform(kw_only_default=True,field_specifiers=(Field,))classModelMetaclass(ABCMeta):@no_type_check# noqa C901def__new__(mcs,name,bases,namespace,**kwargs):# noqa C901fields:Dict[str,ModelField]={}config=BaseConfigvalidators:'ValidatorListDict'={}pre_root_validators,post_root_validators=[],[]private_attributes:Dict[str,ModelPrivateAttr]={}base_private_attributes:Dict[str,ModelPrivateAttr]={}slots:SetStr=namespace.get('__slots__',())slots={slots}ifisinstance(slots,str)elseset(slots)class_vars:SetStr=set()hash_func:Optional[Callable[[Any],int]]=Noneforbaseinreversed(bases):if_is_base_model_class_definedandissubclass(base,BaseModel)andbase!=BaseModel:fields.update(smart_deepcopy(base.__fields__))config=inherit_config(base.__config__,config)validators=inherit_validators(base.__validators__,validators)pre_root_validators+=base.__pre_root_validators__post_root_validators+=base.__post_root_validators__base_private_attributes.update(base.__private_attributes__)class_vars.update(base.__class_vars__)hash_func=base.__hash__resolve_forward_refs=kwargs.pop('__resolve_forward_refs__',True)allowed_config_kwargs:SetStr={keyforkeyindir(config)ifnot(key.startswith('__')andkey.endswith('__'))# skip dunder methods and attributes}config_kwargs={key:kwargs.pop(key)forkeyinkwargs.keys()&allowed_config_kwargs}config_from_namespace=namespace.get('Config')ifconfig_kwargsandconfig_from_namespace:raiseTypeError('Specifying config in two places is ambiguous, use either Config attribute or class kwargs')config=inherit_config(config_from_namespace,config,**config_kwargs)validators=inherit_validators(extract_validators(namespace),validators)vg=ValidatorGroup(validators)forfinfields.values():f.set_config(config)extra_validators=vg.get_validators(f.name)ifextra_validators:f.class_validators.update(extra_validators)# re-run prepare to add extra validatorsf.populate_validators()prepare_config(config,name)untouched_types=ANNOTATED_FIELD_UNTOUCHED_TYPESdefis_untouched(v:Any)->bool:returnisinstance(v,untouched_types)orv.__class__.__name__=='cython_function_or_method'if(namespace.get('__module__'),namespace.get('__qualname__'))!=('pydantic.main','BaseModel'):annotations=resolve_annotations(namespace.get('__annotations__',{}),namespace.get('__module__',None))# annotation only fields need to come first in fieldsforann_name,ann_typeinannotations.items():ifis_classvar(ann_type):class_vars.add(ann_name)elifis_finalvar_with_default_val(ann_type,namespace.get(ann_name,Undefined)):class_vars.add(ann_name)elifis_valid_field(ann_name):validate_field_name(bases,ann_name)value=namespace.get(ann_name,Undefined)allowed_types=get_args(ann_type)ifis_union(get_origin(ann_type))else(ann_type,)if(is_untouched(value)andann_type!=PyObjectandnotany(lenient_issubclass(get_origin(allowed_type),Type)forallowed_typeinallowed_types)):continuefields[ann_name]=ModelField.infer(name=ann_name,value=value,annotation=ann_type,class_validators=vg.get_validators(ann_name),config=config,)elifann_namenotinnamespaceandconfig.underscore_attrs_are_private:private_attributes[ann_name]=PrivateAttr()untouched_types=UNTOUCHED_TYPES+config.keep_untouchedforvar_name,valueinnamespace.items():can_be_changed=var_namenotinclass_varsandnotis_untouched(value)ifisinstance(value,ModelPrivateAttr):ifnotis_valid_private_name(var_name):raiseNameError(f'Private attributes "{var_name}" must not be a valid field name; 'f'Use sunder or dunder names, e. g. "_{var_name}" or "__{var_name}__"')private_attributes[var_name]=valueelifconfig.underscore_attrs_are_privateandis_valid_private_name(var_name)andcan_be_changed:private_attributes[var_name]=PrivateAttr(default=value)elifis_valid_field(var_name)andvar_namenotinannotationsandcan_be_changed:validate_field_name(bases,var_name)inferred=ModelField.infer(name=var_name,value=value,annotation=annotations.get(var_name,Undefined),class_validators=vg.get_validators(var_name),config=config,)ifvar_nameinfields:iflenient_issubclass(inferred.type_,fields[var_name].type_):inferred.type_=fields[var_name].type_else:raiseTypeError(f'The type of {name}.{var_name} differs from the new default value; 'f'if you wish to change the type of this field, please use a type annotation')fields[var_name]=inferred_custom_root_type=ROOT_KEYinfieldsif_custom_root_type:validate_custom_root_type(fields)vg.check_for_unused()ifconfig.json_encoders:json_encoder=partial(custom_pydantic_encoder,config.json_encoders)else:json_encoder=pydantic_encoderpre_rv_new,post_rv_new=extract_root_validators(namespace)ifhash_funcisNone:hash_func=generate_hash_function(config.frozen)exclude_from_namespace=fields|private_attributes.keys()|{'__slots__'}new_namespace={'__config__':config,'__fields__':fields,'__exclude_fields__':{name:field.field_info.excludeforname,fieldinfields.items()iffield.field_info.excludeisnotNone}orNone,'__include_fields__':{name:field.field_info.includeforname,fieldinfields.items()iffield.field_info.includeisnotNone}orNone,'__validators__':vg.validators,'__pre_root_validators__':unique_list(pre_root_validators+pre_rv_new,name_factory=lambdav:v.__name__,),'__post_root_validators__':unique_list(post_root_validators+post_rv_new,name_factory=lambdaskip_on_failure_and_v:skip_on_failure_and_v[1].__name__,),'__schema_cache__':{},'__json_encoder__':staticmethod(json_encoder),'__custom_root_type__':_custom_root_type,'__private_attributes__':{**base_private_attributes,**private_attributes},'__slots__':slots|private_attributes.keys(),'__hash__':hash_func,'__class_vars__':class_vars,**{n:vforn,vinnamespace.items()ifnnotinexclude_from_namespace},}cls=super().__new__(mcs,name,bases,new_namespace,**kwargs)# set __signature__ attr only for model class, but not for its instancescls.__signature__=ClassAttribute('__signature__',generate_model_signature(cls.__init__,fields,config))ifresolve_forward_refs:cls.__try_update_forward_refs__()# preserve `__set_name__` protocol defined in https://peps.python.org/pep-0487# for attributes not in `new_namespace` (e.g. private attributes)forname,objinnamespace.items():ifnamenotinnew_namespace:set_name=getattr(obj,'__set_name__',None)ifcallable(set_name):set_name(cls,name)returnclsdef__instancecheck__(self,instance:Any)->bool:""" Avoid calling ABC _abc_subclasscheck unless we're pretty sure. See #3829 and python/cpython#92810 """returnhasattr(instance,'__post_root_validators__')andsuper().__instancecheck__(instance)object_setattr=object.__setattr__classBaseModel(Representation,metaclass=ModelMetaclass):ifTYPE_CHECKING:# populated by the metaclass, defined here to help IDEs only__fields__:ClassVar[Dict[str,ModelField]]={}__include_fields__:ClassVar[Optional[Mapping[str,Any]]]=None__exclude_fields__:ClassVar[Optional[Mapping[str,Any]]]=None__validators__:ClassVar[Dict[str,AnyCallable]]={}__pre_root_validators__:ClassVar[List[AnyCallable]]__post_root_validators__:ClassVar[List[Tuple[bool,AnyCallable]]]__config__:ClassVar[Type[BaseConfig]]=BaseConfig__json_encoder__:ClassVar[Callable[[Any],Any]]=lambdax:x__schema_cache__:ClassVar['DictAny']={}__custom_root_type__:ClassVar[bool]=False__signature__:ClassVar['Signature']__private_attributes__:ClassVar[Dict[str,ModelPrivateAttr]]__class_vars__:ClassVar[SetStr]__fields_set__:ClassVar[SetStr]=set()Config=BaseConfig__slots__=('__dict__','__fields_set__')__doc__=''# Null out the Representation docstringdef__init__(__pydantic_self__,**data:Any)->None:""" Create a new model by parsing and validating input data from keyword arguments. Raises ValidationError if the input data cannot be parsed to form a valid model. """# Uses something other than `self` the first arg to allow "self" as a settable attributevalues,fields_set,validation_error=validate_model(__pydantic_self__.__class__,data)ifvalidation_error:raisevalidation_errortry:object_setattr(__pydantic_self__,'__dict__',values)exceptTypeErrorase:raiseTypeError('Model values must be a dict; you may not have returned a dictionary from a root validator')fromeobject_setattr(__pydantic_self__,'__fields_set__',fields_set)__pydantic_self__._init_private_attributes()@no_type_checkdef__setattr__(self,name,value):# noqa: C901 (ignore complexity)ifnameinself.__private_attributes__ornameinDUNDER_ATTRIBUTES:returnobject_setattr(self,name,value)ifself.__config__.extraisnotExtra.allowandnamenotinself.__fields__:raiseValueError(f'"{self.__class__.__name__}" object has no field "{name}"')elifnotself.__config__.allow_mutationorself.__config__.frozen:raiseTypeError(f'"{self.__class__.__name__}" is immutable and does not support item assignment')elifnameinself.__fields__andself.__fields__[name].final:raiseTypeError(f'"{self.__class__.__name__}" object "{name}" field is final and does not support reassignment')elifself.__config__.validate_assignment:new_values={**self.__dict__,name:value}forvalidatorinself.__pre_root_validators__:try:new_values=validator(self.__class__,new_values)except(ValueError,TypeError,AssertionError)asexc:raiseValidationError([ErrorWrapper(exc,loc=ROOT_KEY)],self.__class__)known_field=self.__fields__.get(name,None)ifknown_field:# We want to# - make sure validators are called without the current value for this field inside `values`# - keep other values (e.g. submodels) untouched (using `BaseModel.dict()` will change them into dicts)# - keep the order of the fieldsifnotknown_field.field_info.allow_mutation:raiseTypeError(f'"{known_field.name}" has allow_mutation set to False and cannot be assigned')dict_without_original_value={k:vfork,vinself.__dict__.items()ifk!=name}value,error_=known_field.validate(value,dict_without_original_value,loc=name,cls=self.__class__)iferror_:raiseValidationError([error_],self.__class__)else:new_values[name]=valueerrors=[]forskip_on_failure,validatorinself.__post_root_validators__:ifskip_on_failureanderrors:continuetry:new_values=validator(self.__class__,new_values)except(ValueError,TypeError,AssertionError)asexc:errors.append(ErrorWrapper(exc,loc=ROOT_KEY))iferrors:raiseValidationError(errors,self.__class__)# update the whole __dict__ as other values than just `value`# may be changed (e.g. with `root_validator`)object_setattr(self,'__dict__',new_values)else:self.__dict__[name]=valueself.__fields_set__.add(name)def__getstate__(self)->'DictAny':private_attrs=((k,getattr(self,k,Undefined))forkinself.__private_attributes__)return{'__dict__':self.__dict__,'__fields_set__':self.__fields_set__,'__private_attribute_values__':{k:vfork,vinprivate_attrsifvisnotUndefined},}def__setstate__(self,state:'DictAny')->None:object_setattr(self,'__dict__',state['__dict__'])object_setattr(self,'__fields_set__',state['__fields_set__'])forname,valueinstate.get('__private_attribute_values__',{}).items():object_setattr(self,name,value)def_init_private_attributes(self)->None:forname,private_attrinself.__private_attributes__.items():default=private_attr.get_default()ifdefaultisnotUndefined:object_setattr(self,name,default)defdict(self,*,include:Optional[Union['AbstractSetIntStr','MappingIntStrAny']]=None,exclude:Optional[Union['AbstractSetIntStr','MappingIntStrAny']]=None,by_alias:bool=False,skip_defaults:Optional[bool]=None,exclude_unset:bool=False,exclude_defaults:bool=False,exclude_none:bool=False,)->'DictStrAny':""" Generate a dictionary representation of the model, optionally specifying which fields to include or exclude. """ifskip_defaultsisnotNone:warnings.warn(f'{self.__class__.__name__}.dict(): "skip_defaults" is deprecated and replaced by "exclude_unset"',DeprecationWarning,)exclude_unset=skip_defaultsreturndict(self._iter(to_dict=True,by_alias=by_alias,include=include,exclude=exclude,exclude_unset=exclude_unset,exclude_defaults=exclude_defaults,exclude_none=exclude_none,))defjson(self,*,include:Optional[Union['AbstractSetIntStr','MappingIntStrAny']]=None,exclude:Optional[Union['AbstractSetIntStr','MappingIntStrAny']]=None,by_alias:bool=False,skip_defaults:Optional[bool]=None,exclude_unset:bool=False,exclude_defaults:bool=False,exclude_none:bool=False,encoder:Optional[Callable[[Any],Any]]=None,models_as_dict:bool=True,**dumps_kwargs:Any,)->str:""" Generate a JSON representation of the model, `include` and `exclude` arguments as per `dict()`. `encoder` is an optional function to supply as `default` to json.dumps(), other arguments as per `json.dumps()`. """ifskip_defaultsisnotNone:warnings.warn(f'{self.__class__.__name__}.json(): "skip_defaults" is deprecated and replaced by "exclude_unset"',DeprecationWarning,)exclude_unset=skip_defaultsencoder=cast(Callable[[Any],Any],encoderorself.__json_encoder__)# We don't directly call `self.dict()`, which does exactly this with `to_dict=True`# because we want to be able to keep raw `BaseModel` instances and not as `dict`.# This allows users to write custom JSON encoders for given `BaseModel` classes.data=dict(self._iter(to_dict=models_as_dict,by_alias=by_alias,include=include,exclude=exclude,exclude_unset=exclude_unset,exclude_defaults=exclude_defaults,exclude_none=exclude_none,))ifself.__custom_root_type__:data=data[ROOT_KEY]returnself.__config__.json_dumps(data,default=encoder,**dumps_kwargs)@classmethoddef_enforce_dict_if_root(cls,obj:Any)->Any:ifcls.__custom_root_type__and(not(isinstance(obj,dict)andobj.keys()=={ROOT_KEY})andnot(isinstance(obj,BaseModel)andobj.__fields__.keys()=={ROOT_KEY})orcls.__fields__[ROOT_KEY].shapeinMAPPING_LIKE_SHAPES):return{ROOT_KEY:obj}else:returnobj@classmethoddefparse_obj(cls:Type['Model'],obj:Any)->'Model':obj=cls._enforce_dict_if_root(obj)ifnotisinstance(obj,dict):try:obj=dict(obj)except(TypeError,ValueError)ase:exc=TypeError(f'{cls.__name__} expected dict not {obj.__class__.__name__}')raiseValidationError([ErrorWrapper(exc,loc=ROOT_KEY)],cls)fromereturncls(**obj)@classmethoddefparse_raw(cls:Type['Model'],b:StrBytes,*,content_type:str=None,encoding:str='utf8',proto:Protocol=None,allow_pickle:bool=False,)->'Model':try:obj=load_str_bytes(b,proto=proto,content_type=content_type,encoding=encoding,allow_pickle=allow_pickle,json_loads=cls.__config__.json_loads,)except(ValueError,TypeError,UnicodeDecodeError)ase:raiseValidationError([ErrorWrapper(e,loc=ROOT_KEY)],cls)returncls.parse_obj(obj)@classmethoddefparse_file(cls:Type['Model'],path:Union[str,Path],*,content_type:str=None,encoding:str='utf8',proto:Protocol=None,allow_pickle:bool=False,)->'Model':obj=load_file(path,proto=proto,content_type=content_type,encoding=encoding,allow_pickle=allow_pickle,json_loads=cls.__config__.json_loads,)returncls.parse_obj(obj)@classmethoddeffrom_orm(cls:Type['Model'],obj:Any)->'Model':ifnotcls.__config__.orm_mode:raiseConfigError('You must have the config attribute orm_mode=True to use from_orm')obj={ROOT_KEY:obj}ifcls.__custom_root_type__elsecls._decompose_class(obj)m=cls.__new__(cls)values,fields_set,validation_error=validate_model(cls,obj)ifvalidation_error:raisevalidation_errorobject_setattr(m,'__dict__',values)object_setattr(m,'__fields_set__',fields_set)m._init_private_attributes()returnm@classmethoddefconstruct(cls:Type['Model'],_fields_set:Optional['SetStr']=None,**values:Any)->'Model':""" Creates a new model setting __dict__ and __fields_set__ from trusted or pre-validated data. Default values are respected, but no other validation is performed. Behaves as if `Config.extra = 'allow'` was set since it adds all passed values """m=cls.__new__(cls)fields_values:Dict[str,Any]={}forname,fieldincls.__fields__.items():iffield.alt_aliasandfield.aliasinvalues:fields_values[name]=values[field.alias]elifnameinvalues:fields_values[name]=values[name]elifnotfield.required:fields_values[name]=field.get_default()fields_values.update(values)object_setattr(m,'__dict__',fields_values)if_fields_setisNone:_fields_set=set(values.keys())object_setattr(m,'__fields_set__',_fields_set)m._init_private_attributes()returnmdef_copy_and_set_values(self:'Model',values:'DictStrAny',fields_set:'SetStr',*,deep:bool)->'Model':ifdeep:# chances of having empty dict here are quite low for using smart_deepcopyvalues=deepcopy(values)cls=self.__class__m=cls.__new__(cls)object_setattr(m,'__dict__',values)object_setattr(m,'__fields_set__',fields_set)fornameinself.__private_attributes__:value=getattr(self,name,Undefined)ifvalueisnotUndefined:ifdeep:value=deepcopy(value)object_setattr(m,name,value)returnmdefcopy(self:'Model',*,include:Optional[Union['AbstractSetIntStr','MappingIntStrAny']]=None,exclude:Optional[Union['AbstractSetIntStr','MappingIntStrAny']]=None,update:Optional['DictStrAny']=None,deep:bool=False,)->'Model':""" Duplicate a model, optionally choose which fields to include, exclude and change. :param include: fields to include in new model :param exclude: fields to exclude from new model, as with values this takes precedence over include :param update: values to change/add in the new model. Note: the data is not validated before creating the new model: you should trust this data :param deep: set to `True` to make a deep copy of the model :return: new model instance """values=dict(self._iter(to_dict=False,by_alias=False,include=include,exclude=exclude,exclude_unset=False),**(updateor{}),)# new `__fields_set__` can have unset optional fields with a set value in `update` kwargifupdate:fields_set=self.__fields_set__|update.keys()else:fields_set=set(self.__fields_set__)returnself._copy_and_set_values(values,fields_set,deep=deep)@classmethoddefschema(cls,by_alias:bool=True,ref_template:str=default_ref_template)->'DictStrAny':cached=cls.__schema_cache__.get((by_alias,ref_template))ifcachedisnotNone:returncacheds=model_schema(cls,by_alias=by_alias,ref_template=ref_template)cls.__schema_cache__[(by_alias,ref_template)]=sreturns@classmethoddefschema_json(cls,*,by_alias:bool=True,ref_template:str=default_ref_template,**dumps_kwargs:Any)->str:frompydantic.v1.jsonimportpydantic_encoderreturncls.__config__.json_dumps(cls.schema(by_alias=by_alias,ref_template=ref_template),default=pydantic_encoder,**dumps_kwargs)@classmethoddef__get_validators__(cls)->'CallableGenerator':yieldcls.validate@classmethoddefvalidate(cls:Type['Model'],value:Any)->'Model':ifisinstance(value,cls):copy_on_model_validation=cls.__config__.copy_on_model_validation# whether to deep or shallow copy the model on validation, None means do not copydeep_copy:Optional[bool]=Noneifcopy_on_model_validationnotin{'deep','shallow','none'}:# Warn about deprecated behaviorwarnings.warn("`copy_on_model_validation` should be a string: 'deep', 'shallow' or 'none'",DeprecationWarning)ifcopy_on_model_validation:deep_copy=Falseifcopy_on_model_validation=='shallow':# shallow copydeep_copy=Falseelifcopy_on_model_validation=='deep':# deep copydeep_copy=Trueifdeep_copyisNone:returnvalueelse:returnvalue._copy_and_set_values(value.__dict__,value.__fields_set__,deep=deep_copy)value=cls._enforce_dict_if_root(value)ifisinstance(value,dict):returncls(**value)elifcls.__config__.orm_mode:returncls.from_orm(value)else:try:value_as_dict=dict(value)except(TypeError,ValueError)ase:raiseDictError()fromereturncls(**value_as_dict)@classmethoddef_decompose_class(cls:Type['Model'],obj:Any)->GetterDict:ifisinstance(obj,GetterDict):returnobjreturncls.__config__.getter_dict(obj)@classmethod@no_type_checkdef_get_value(cls,v:Any,to_dict:bool,by_alias:bool,include:Optional[Union['AbstractSetIntStr','MappingIntStrAny']],exclude:Optional[Union['AbstractSetIntStr','MappingIntStrAny']],exclude_unset:bool,exclude_defaults:bool,exclude_none:bool,)->Any:ifisinstance(v,BaseModel):ifto_dict:v_dict=v.dict(by_alias=by_alias,exclude_unset=exclude_unset,exclude_defaults=exclude_defaults,include=include,exclude=exclude,exclude_none=exclude_none,)ifROOT_KEYinv_dict:returnv_dict[ROOT_KEY]returnv_dictelse:returnv.copy(include=include,exclude=exclude)value_exclude=ValueItems(v,exclude)ifexcludeelseNonevalue_include=ValueItems(v,include)ifincludeelseNoneifisinstance(v,dict):return{k_:cls._get_value(v_,to_dict=to_dict,by_alias=by_alias,exclude_unset=exclude_unset,exclude_defaults=exclude_defaults,include=value_includeandvalue_include.for_element(k_),exclude=value_excludeandvalue_exclude.for_element(k_),exclude_none=exclude_none,)fork_,v_inv.items()if(notvalue_excludeornotvalue_exclude.is_excluded(k_))and(notvalue_includeorvalue_include.is_included(k_))}elifsequence_like(v):seq_args=(cls._get_value(v_,to_dict=to_dict,by_alias=by_alias,exclude_unset=exclude_unset,exclude_defaults=exclude_defaults,include=value_includeandvalue_include.for_element(i),exclude=value_excludeandvalue_exclude.for_element(i),exclude_none=exclude_none,)fori,v_inenumerate(v)if(notvalue_excludeornotvalue_exclude.is_excluded(i))and(notvalue_includeorvalue_include.is_included(i)))returnv.__class__(*seq_args)ifis_namedtuple(v.__class__)elsev.__class__(seq_args)elifisinstance(v,Enum)andgetattr(cls.Config,'use_enum_values',False):returnv.valueelse:returnv@classmethoddef__try_update_forward_refs__(cls,**localns:Any)->None:""" Same as update_forward_refs but will not raise exception when forward references are not defined. """update_model_forward_refs(cls,cls.__fields__.values(),cls.__config__.json_encoders,localns,(NameError,))@classmethoddefupdate_forward_refs(cls,**localns:Any)->None:""" Try to update ForwardRefs on fields based on this Model, globalns and localns. """update_model_forward_refs(cls,cls.__fields__.values(),cls.__config__.json_encoders,localns)def__iter__(self)->'TupleGenerator':""" so `dict(model)` works """yield fromself.__dict__.items()def_iter(self,to_dict:bool=False,by_alias:bool=False,include:Optional[Union['AbstractSetIntStr','MappingIntStrAny']]=None,exclude:Optional[Union['AbstractSetIntStr','MappingIntStrAny']]=None,exclude_unset:bool=False,exclude_defaults:bool=False,exclude_none:bool=False,)->'TupleGenerator':# Merge field set excludes with explicit exclude parameter with explicit overriding field set options.# The extra "is not None" guards are not logically necessary but optimizes performance for the simple case.ifexcludeisnotNoneorself.__exclude_fields__isnotNone:exclude=ValueItems.merge(self.__exclude_fields__,exclude)ifincludeisnotNoneorself.__include_fields__isnotNone:include=ValueItems.merge(self.__include_fields__,include,intersect=True)allowed_keys=self._calculate_keys(include=include,exclude=exclude,exclude_unset=exclude_unset# type: ignore)ifallowed_keysisNoneandnot(to_dictorby_aliasorexclude_unsetorexclude_defaultsorexclude_none):# huge boost for plain _iter()yield fromself.__dict__.items()returnvalue_exclude=ValueItems(self,exclude)ifexcludeisnotNoneelseNonevalue_include=ValueItems(self,include)ifincludeisnotNoneelseNoneforfield_key,vinself.__dict__.items():if(allowed_keysisnotNoneandfield_keynotinallowed_keys)or(exclude_noneandvisNone):continueifexclude_defaults:model_field=self.__fields__.get(field_key)ifnotgetattr(model_field,'required',True)andgetattr(model_field,'default',_missing)==v:continueifby_aliasandfield_keyinself.__fields__:dict_key=self.__fields__[field_key].aliaselse:dict_key=field_keyifto_dictorvalue_includeorvalue_exclude:v=self._get_value(v,to_dict=to_dict,by_alias=by_alias,include=value_includeandvalue_include.for_element(field_key),exclude=value_excludeandvalue_exclude.for_element(field_key),exclude_unset=exclude_unset,exclude_defaults=exclude_defaults,exclude_none=exclude_none,)yielddict_key,vdef_calculate_keys(self,include:Optional['MappingIntStrAny'],exclude:Optional['MappingIntStrAny'],exclude_unset:bool,update:Optional['DictStrAny']=None,)->Optional[AbstractSet[str]]:ifincludeisNoneandexcludeisNoneandexclude_unsetisFalse:returnNonekeys:AbstractSet[str]ifexclude_unset:keys=self.__fields_set__.copy()else:keys=self.__dict__.keys()ifincludeisnotNone:keys&=include.keys()ifupdate:keys-=update.keys()ifexclude:keys-={kfork,vinexclude.items()ifValueItems.is_true(v)}returnkeysdef__eq__(self,other:Any)->bool:ifisinstance(other,BaseModel):returnself.dict()==other.dict()else:returnself.dict()==otherdef__repr_args__(self)->'ReprArgs':return[(k,v)fork,vinself.__dict__.items()ifknotinDUNDER_ATTRIBUTESand(knotinself.__fields__orself.__fields__[k].field_info.repr)]_is_base_model_class_defined=True@overloaddefcreate_model(__model_name:str,*,__config__:Optional[Type[BaseConfig]]=None,__base__:None=None,__module__:str=__name__,__validators__:Dict[str,'AnyClassMethod']=None,__cls_kwargs__:Dict[str,Any]=None,**field_definitions:Any,)->Type['BaseModel']:...@overloaddefcreate_model(__model_name:str,*,__config__:Optional[Type[BaseConfig]]=None,__base__:Union[Type['Model'],Tuple[Type['Model'],...]],__module__:str=__name__,__validators__:Dict[str,'AnyClassMethod']=None,__cls_kwargs__:Dict[str,Any]=None,**field_definitions:Any,)->Type['Model']:...defcreate_model(__model_name:str,*,__config__:Optional[Type[BaseConfig]]=None,__base__:Union[None,Type['Model'],Tuple[Type['Model'],...]]=None,__module__:str=__name__,__validators__:Dict[str,'AnyClassMethod']=None,__cls_kwargs__:Dict[str,Any]=None,__slots__:Optional[Tuple[str,...]]=None,**field_definitions:Any,)->Type['Model']:""" Dynamically create a model. :param __model_name: name of the created model :param __config__: config class to use for the new model :param __base__: base class for the new model to inherit from :param __module__: module of the created model :param __validators__: a dict of method names and @validator class methods :param __cls_kwargs__: a dict for class creation :param __slots__: Deprecated, `__slots__` should not be passed to `create_model` :param field_definitions: fields of the model (or extra fields if a base is supplied) in the format `<name>=(<type>, <default default>)` or `<name>=<default value>, e.g. `foobar=(str, ...)` or `foobar=123`, or, for complex use-cases, in the format `<name>=<Field>` or `<name>=(<type>, <FieldInfo>)`, e.g. `foo=Field(datetime, default_factory=datetime.utcnow, alias='bar')` or `foo=(str, FieldInfo(title='Foo'))` """if__slots__isnotNone:# __slots__ will be ignored from here onwarnings.warn('__slots__ should not be passed to create_model',RuntimeWarning)if__base__isnotNone:if__config__isnotNone:raiseConfigError('to avoid confusion __config__ and __base__ cannot be used together')ifnotisinstance(__base__,tuple):__base__=(__base__,)else:__base__=(cast(Type['Model'],BaseModel),)__cls_kwargs__=__cls_kwargs__or{}fields={}annotations={}forf_name,f_definfield_definitions.items():ifnotis_valid_field(f_name):warnings.warn(f'fields may not start with an underscore, ignoring "{f_name}"',RuntimeWarning)ifisinstance(f_def,tuple):try:f_annotation,f_value=f_defexceptValueErrorase:raiseConfigError('field definitions should either be a tuple of (<type>, <default>) or just a ''default value, unfortunately this means tuples as ''default values are not allowed')fromeelse:f_annotation,f_value=None,f_defiff_annotation:annotations[f_name]=f_annotationfields[f_name]=f_valuenamespace:'DictStrAny'={'__annotations__':annotations,'__module__':__module__}if__validators__:namespace.update(__validators__)namespace.update(fields)if__config__:namespace['Config']=inherit_config(__config__,BaseConfig)resolved_bases=resolve_bases(__base__)meta,ns,kwds=prepare_class(__model_name,resolved_bases,kwds=__cls_kwargs__)ifresolved_basesisnot__base__:ns['__orig_bases__']=__base__namespace.update(ns)returnmeta(__model_name,resolved_bases,namespace,**kwds)_missing=object()defvalidate_model(# noqa: C901 (ignore complexity)model:Type[BaseModel],input_data:'DictStrAny',cls:'ModelOrDc'=None)->Tuple['DictStrAny','SetStr',Optional[ValidationError]]:""" validate data against a model. """values={}errors=[]# input_data names, possibly aliasnames_used=set()# field names, never aliasesfields_set=set()config=model.__config__check_extra=config.extraisnotExtra.ignorecls_=clsormodelforvalidatorinmodel.__pre_root_validators__:try:input_data=validator(cls_,input_data)except(ValueError,TypeError,AssertionError)asexc:return{},set(),ValidationError([ErrorWrapper(exc,loc=ROOT_KEY)],cls_)forname,fieldinmodel.__fields__.items():value=input_data.get(field.alias,_missing)using_name=Falseifvalueis_missingandconfig.allow_population_by_field_nameandfield.alt_alias:value=input_data.get(field.name,_missing)using_name=Trueifvalueis_missing:iffield.required:errors.append(ErrorWrapper(MissingError(),loc=field.alias))continuevalue=field.get_default()ifnotconfig.validate_allandnotfield.validate_always:values[name]=valuecontinueelse:fields_set.add(name)ifcheck_extra:names_used.add(field.nameifusing_nameelsefield.alias)v_,errors_=field.validate(value,values,loc=field.alias,cls=cls_)ifisinstance(errors_,ErrorWrapper):errors.append(errors_)elifisinstance(errors_,list):errors.extend(errors_)else:values[name]=v_ifcheck_extra:ifisinstance(input_data,GetterDict):extra=input_data.extra_keys()-names_usedelse:extra=input_data.keys()-names_usedifextra:fields_set|=extraifconfig.extraisExtra.allow:forfinextra:values[f]=input_data[f]else:forfinsorted(extra):errors.append(ErrorWrapper(ExtraError(),loc=f))forskip_on_failure,validatorinmodel.__post_root_validators__:ifskip_on_failureanderrors:continuetry:values=validator(cls_,values)except(ValueError,TypeError,AssertionError)asexc:errors.append(ErrorWrapper(exc,loc=ROOT_KEY))iferrors:returnvalues,fields_set,ValidationError(errors,cls_)else:returnvalues,fields_set,None