U s\j`@sUddlmZddlZddlZddlZddlmZmZddl m Z ddl m Z ddl mZddlmZmZmZmZmZmZmZmZddlmZdd lmZmZmZmZdd lm Z dd l!m"Z"m#Z#dd l$m%Z%dd l&m'Z'm(Z(ddl)m*Z*erddl+m,Z,eee-eeee-feeee-fdffZ.edZ/de0d<Gddde1Z2GdddeZ3Gddde3Z4Gddde3Z5Gddde5Z6Gddde5Z7Gd d!d!e7Z8dd"d#d$d%d&d'd(d)d*Z9d+d,d&d-d.d/Z:d+d&d0d1d2Z;dS)3) annotationsN)ABCabstractmethod)deque) is_dataclass)Path) TYPE_CHECKINGAnyListMappingSequenceTupleUnioncast) dotenv_values) AliasChoices AliasPath BaseModelJson)origin_is_union) deep_updatelenient_issubclass) FieldInfo)get_args get_origin)path_type_label) BaseSettings. DotenvTypeENV_FILE_SENTINELc@s eZdZdS) SettingsErrorN)__name__ __module__ __qualname__r$r$P/root/pica-gateway/venv/lib/python3.8/site-packages/pydantic_settings/sources.pyr !sr c@seZdZdZddddZedddd d d Zdd d ddZdddd ddddZdddddddZ eddddZ dS)PydanticBaseSettingsSourcezi Abstract base class for settings sources, every settings source classes should inherit from it. type[BaseSettings]) settings_clscCs||_|j|_dSN)r( model_configconfig)selfr(r$r$r%__init__*sz#PydanticBaseSettingsSource.__init__rstrtuple[Any, str, bool]field field_namereturncCsdS)a Gets the value, the key for model creation, and a flag to determine whether value is complex. This is an abstract method that should be overridden in every settings source classes. Args: field: The field. field_name: The field name. Returns: A tuple contains the key, value and a flag to determine whether value is complex. Nr$r,r1r2r$r$r%get_field_value.sz*PydanticBaseSettingsSource.get_field_valueboolr1r3cCst|j|jS)z Checks whether a field is complex, in which case it will attempt to be parsed as JSON. Args: field: The field. Returns: Whether the field is complex. )_annotation_is_complex annotationmetadata)r,r1r$r$r%field_is_complex>s z+PydanticBaseSettingsSource.field_is_complexr r2r1valuevalue_is_complexr3cCs(|dk r$||s|r$||||S|S)aF Prepares the value of a field. Args: field_name: The field name. field: The field. value: The value of the field that has to be prepared. value_is_complex: A flag to determine whether value is complex. Returns: The prepared value. N)r;decode_complex_value)r,r2r1r=r>r$r$r%prepare_field_valueJs z.PydanticBaseSettingsSource.prepare_field_value)r2r1r=r3cCs t|S)a Decode the value for a complex field Args: field_name: The field name. field: The field. value: The value of the field that has to be prepared. Returns: The decoded value for further preparation )jsonloads)r,r2r1r=r$r$r%r?[s z/PydanticBaseSettingsSource.decode_complex_valuedict[str, Any]r3cCsdSr)r$r,r$r$r%__call__isz#PydanticBaseSettingsSource.__call__N) r!r"r#__doc__r-rr5r;r@r?rFr$r$r$r%r&%s r&csVeZdZdZdddfdd Zddd d d d Zdd ddZdd ddZZS)InitSettingsSourcezX Source class for loading values provided during settings class initialization. r'rC)r( init_kwargscs||_t|dSr))rIsuperr-)r,r(rI __class__r$r%r-sszInitSettingsSource.__init__rr.r/r0cCsdS)N)NrFr$r4r$r$r%r5wsz"InitSettingsSource.get_field_valuerDcCs|jSr)rIrEr$r$r%rF{szInitSettingsSource.__call__cCsd|jdS)NzInitSettingsSource(init_kwargs=)rMrEr$r$r%__repr__~szInitSettingsSource.__repr__) r!r"r#rGr-r5rFrO __classcell__r$r$rKr%rHns rHcsleZdZddddddfdd Zd d d d d Zd d ddddZd dddddZddddZZS)PydanticBaseEnvSettingsSourceNr' bool | None str | NoneNone)r(case_sensitive env_prefixr3csHt||dk r|n |jdd|_|dk r4|n |jdd|_dS)NrUFrVr)rJr-r+getrUrV)r,r(rUrVrKr$r%r-s z&PydanticBaseEnvSettingsSource.__init__r.)r=r3cCs|js|S|Sr))rUlower)r,r=r$r$r%_apply_case_sensitivesz3PydanticBaseEnvSettingsSource._apply_case_sensitiverzlist[tuple[str, str, bool]]r0cCsg}t|jttfr |j}n|j}|rt|tr|D]t}t|trn||||t |dkrddndfq8t|tr8t t|d}||||t |dkrdndfq8q||||dfn||||j |df|S)a Extracts field info. This info is used to get the value of field from environment variables. It returns a list of tuples, each tuple contains: * field_key: The key of field that has to be used in model creation. * env_name: The environment variable name of the field. * value_is_complex: A flag to determine whether the value from environment variable is complex and has to be parsed. Args: field (FieldInfo): The field. field_name (str): The field name. Returns: list[tuple[str, str, bool]]: List of tuples, each tuple contains field_key, env_name, and value_is_complex. TFr) isinstancevalidation_aliasrrZconvert_to_aliaseslistr.appendrYlenrrV)r,r1r2Z field_infoZv_aliasaliasZ first_argr$r$r%_extract_field_infos"   ( z1PydanticBaseEnvSettingsSource._extract_field_inforC)r1 field_valuesr3c Csi}|D]\}}d}|jr*t|jds4|||<q |jjD]&\}}|js@||kr@|}qhq@|sv|||<q t|jtrt|t r| ||||<q |||<q |S)ae Replace field names in values dict by looking in models fields insensitively. By having the following models: ```py class SubSubSub(BaseModel): VaL3: str class SubSub(BaseModel): Val2: str SUB_sub_SuB: SubSubSub class Sub(BaseModel): VAL1: str SUB_sub: SubSub class Settings(BaseSettings): nested: Sub model_config = SettingsConfigDict(env_nested_delimiter='__') ``` Then: _replace_field_names_case_insensitively( field, {"val1": "v1", "sub_SUB": {"VAL2": "v2", "sub_SUB_sUb": {"vAl3": "v3"}}} ) Returns {'VAL1': 'v1', 'SUB_sub': {'Val2': 'v2', 'SUB_sub_SuB': {'VaL3': 'v3'}}} N model_fields) itemsr9hasattrrcr\rXrrr[dict'_replace_field_names_case_insensitively) r,r1rbvaluesnamer=Zsub_model_fieldZsub_model_field_namefr$r$r%rgs" zEPydanticBaseEnvSettingsSource._replace_field_names_case_insensitivelyrDc Csi}|jjD]\}}z|||\}}}Wn>tk rl}z td|d|jjd|W5d}~XYnXz|||||}Wn>t k r}z td|d|jjd|W5d}~XYnX|dk r|j st |j t rt|tr|||||<q|||<q|S)Nzerror getting value for field "z" from source ""zerror parsing value for field ")r(rcrdr5 Exceptionr rLr!r@ ValueErrorrUrr9rr[rfrg)r,datar2r1Z field_value field_keyr>er$r$r%rFs8  z&PydanticBaseEnvSettingsSource.__call__)NN) r!r"r#r-rYrargrFrPr$r$rKr%rQs(;rQcszeZdZdZdddddddfd d Zd d fd d ZedddddddZddddddZdd ddZ Z S)SecretsSettingsSourcezE Source class for loading settings values from secret files. Nr'zstr | Path | NonerRrSrT)r( secrets_dirrUrVr3cs.t||||dk r|n |jd|_dS)Nrr)rJr-r+rWrr)r,r(rrrUrVrKr$r%r-szSecretsSettingsSource.__init__rCrDcsli}|jdkr|St|j|_|jsDtd|jd|S|jsbtdt |jt S)z4 Build fields from "secrets" files. Nz directory "z" does not existz.secrets_dir must reference a directory, not a ) rrr expanduser secrets_pathexistswarningswarnis_dirr rrJrF)r,ZsecretsrKr$r%rFs   zSecretsSettingsSource.__call__rr.r6z Path | None)dir_path file_namerUr3cCsB|D]4}|j|kr|S|s|j|kr|SqdS)af Find a file within path's directory matching filename, optionally ignoring case. Args: dir_path: Directory path. file_name: File name. case_sensitive: Whether to search for file name case sensitively. Returns: Whether file path or `None` if file does not exist in directory. N)iterdirrirX)clsryrzrUrjr$r$r%find_case_path2s   z$SecretsSettingsSource.find_case_pathrr/r0cCsx|||D]`\}}}||j||j}|s.q |rL|||fStjd|dt |dddq d||fS)ag Gets the value for field from secret file and a flag to determine whether value is complex. Args: field: The field. field_name: The field name. Returns: A tuple contains the key, value if the file exists otherwise `None`, and a flag to determine whether value is complex. zattempted to load secret file "z" but found a z instead.) stacklevelN) rar}rtrUis_file read_textstriprvrwr)r,r1r2roenv_namer>pathr$r$r%r5Fs z%SecretsSettingsSource.get_field_valuecCsd|jdS)Nz"SecretsSettingsSource(secrets_dir=rN)rrrEr$r$r%rOcszSecretsSettingsSource.__repr__)NNN) r!r"r#rGr-rF classmethodr}r5rOrPr$r$rKr%rqs rqcseZdZdZd,ddddddfdd Zd d d d ZddddddZddddddddZddddddZdddd d!Z e d"dd"d#d$d%Z ddd d&d'd(d)Z dd d*d+Z ZS)-EnvSettingsSourcezN Source class for loading settings values from environment variables. Nr'rRrSrT)r(rUrVenv_nested_delimiterr3csDt||||dk r|n |jd|_t|j|_||_ dS)Nr) rJr-r+rWrr_rVenv_prefix_len_load_env_varsenv_vars)r,r(rUrVrrKr$r%r-ls  zEnvSettingsSource.__init__Mapping[str, str | None]rDcCs |jr tjSddtjDS)NcSsi|]\}}||qSr$rX.0kvr$r$r% ~sz4EnvSettingsSource._load_env_vars..)rUosenvironrdrEr$r$r%r{sz EnvSettingsSource._load_env_varsrr.r/r0cCs>d}|||D]"\}}}|j|}|dk rq4q|||fS)aq Gets the value for field from environment variables and a flag to determine whether value is complex. Args: field: The field. field_name: The field name. Returns: A tuple contains the key, value if the file exists otherwise `None`, and a flag to determine whether value is complex. N)rarrW)r,r1r2env_valrorr>r$r$r%r5s  z!EnvSettingsSource.get_field_valuer r6r<c Cs||\}}|s|r|dkr8||||j}|r|Sqz||||}Wn*tk rt}z |sd|W5d}~XYnXt|trt|||||jS|Sn |dk r|SdS)a Prepare value for the field. * Extract value for nested field. * Deserialize value to python object for complex field. Args: field: The field. field_name: The field name. Returns: A tuple contains prepared value for the field. Raises: ValuesError: When There is an error in deserializing value for complex field. N)_field_is_complexexplode_env_varsrr?rmr[rfr) r,r2r1r=r> is_complexallow_parse_failureZ env_val_builtrpr$r$r%r@s  z%EnvSettingsSource.prepare_field_valuetype[Any] | None list[Any]r9r:r3cstfddt|DS)Nc3s|]}t|VqdSr))r8)rargr:r$r% sz6EnvSettingsSource._union_is_complex..)anyr)r,r9r:r$rr%_union_is_complexsz#EnvSettingsSource._union_is_complexztuple[bool, bool]r7cCs@||rd}n(tt|jr4||j|jr4d}ndSd|fS)za Find out if a field is complex, and if so whether JSON errors should be ignored FT)FF)r;rrr9rr:)r,r1rr$r$r%rs  z#EnvSettingsSource._field_is_complexzFieldInfo | None)r1keyr3cCsF|rtt|jrdS|jrBt|jdrB|jj|rB|jj|SdS)a Find the field in a sub model by key(env name) By having the following models: ```py class SubSubModel(BaseSettings): dvals: Dict class SubModel(BaseSettings): vals: list[str] sub_sub_model: SubSubModel class Cfg(BaseSettings): sub_model: SubModel ``` Then: next_field(sub_model, 'vals') Returns the `vals` field of `SubModel` class next_field(sub_model, 'sub_sub_model') Returns `sub_sub_model` field of `SubModel` class Args: field: The field. key: The key (env name). Returns: Field if it finds the next field otherwise `None`. Nrc)rrr9rercrW)r1rr$r$r% next_fields   zEnvSettingsSource.next_fieldrC)r2r1rr3c sfdd||D}i}|D]\}tfdd|DsFq&jd}|j^}} } |} |} | D]} | | } | | i} qt| | } | r|r| \}}|rz | | |}Wn*t k r}z |s|W5d}~XYnX|| | <q&|S)a Process env_vars and extract the values of keys containing env_nested_delimiter into nested dictionaries. This is applied to a single field, hence filtering by env_var prefix. Args: field_name: The field name. field: The field. env_vars: Environment variables. Returns: A dictionaty contains extracted values from nested env values. cs g|]\}}}|jqSr$)r)r_rrEr$r% sz6EnvSettingsSource.explode_env_vars..c3s|]}|VqdSr)) startswith)rprefix)rr$r%rsz5EnvSettingsSource.explode_env_vars..N) rardrrsplitrr setdefaultrr?rm)r,r2r1rprefixesresultrenv_name_without_prefixrkeyslast_keyenv_varZ target_fieldrrZallow_json_failurerpr$)rr,r%rs2     z"EnvSettingsSource.explode_env_varscCsd|jd|jdS)Nz'EnvSettingsSource(env_nested_delimiter=, env_prefix_len=rN)rrrEr$r$r%rO!szEnvSettingsSource.__repr__)NNN)r!r"r#rGr-rr5r@rr staticmethodrrrOrPr$r$rKr%rgs( %.rc s|eZdZdZeddddfddddddddfd d Zd d d dZdd dddZdd fdd Zdd ddZ Z S)DotEnvSettingsSourcezB Source class for loading settings values from env files. Nr'zDotenvType | NonerSrRrT)r(env_fileenv_file_encodingrUrVrr3csJ|tkr |n |jd|_|dk r&|n |jd|_t||||dS)Nrr)rr*rWrrrJr-)r,r(rrrUrVrrKr$r%r--s zDotEnvSettingsSource.__init__rrDcCs ||jSr))_read_env_filesrUrEr$r$r%r<sz#DotEnvSettingsSource._load_env_varsr6)rUr3cCsd|j}|dkriSt|ttjfr(|g}i}|D].}t|}|r0|t ||j |dq0|S)NencodingrU) rr[r.rPathLikerrsrupdate read_env_filer)r,rUZ env_filesZ dotenv_varsrZenv_pathr$r$r%r?s z$DotEnvSettingsSource._read_env_filesrCcst}g}|js&dd|D}|jD]Z\}}||jr0|dk r0||jd}| |j ^}}|rv||ks|s0||kr0|||<q0|S)NcSsg|] }|qSr$r)rxr$r$r%rVsz1DotEnvSettingsSource.__call__..) rJrFrUrrrdrrVrrr)r,rnZdata_lower_keysrZ env_valuerZ first_keyrrKr$r%rFQs   zDotEnvSettingsSource.__call__r.c Cs&d|jd|jd|jd|jd S)NzDotEnvSettingsSource(env_file=z, env_file_encoding=z, env_nested_delimiter=rrN)rrrrrEr$r$r%rOfs$zDotEnvSettingsSource.__repr__) r!r"r#rGrr-rrrFrOrPr$r$rKr%r(s rFrrrSr6r) file_pathrrUr3cCs.t||p dd}|s&dd|DS|SdS)Nutf8)rcSsi|]\}}||qSr$rrr$r$r%rrsz!read_env_file..)rrd)rrrUZ file_varsr$r$r%rmsrrrrcCsBtdd|DrdSt|}t|p@t|p@t|dp@t|dS)Ncss|]}t|tVqdSr))r[r)rmdr$r$r%rxsz)_annotation_is_complex..FZ__pydantic_core_schema__Z__get_pydantic_core_schema__)rr_annotation_is_complex_innerre)r9r:originr$r$r%r8wsr8)r9r3c Cs2t|ttfrdSt|ttttttt fp0t |S)NF) rr.bytesrr r tupleset frozensetrr)r9r$r$r%rs r)< __future__rZ _annotationsrArrvabcrr collectionsr dataclassesrpathlibrtypingrr r r r r rrZdotenvrZpydanticrrrrZ pydantic._internal._typing_extrarZpydantic._internal._utilsrrZpydantic.fieldsrtyping_extensionsrrZpydantic_settings.utilsrZpydantic_settings.mainrr.rr__annotations__rmr r&rHrQrqrrrr8rr$r$r$r%sD    (     ,IXBF