Underground-Digital/Workflow-Engine
0
1from typing import Optional2 3from core.model_runtime.entities.provider_entities import CredentialFormSchema, FormType4 5 6class CommonValidator:7 def _validate_and_filter_credential_form_schemas(8 self, credential_form_schemas: list[CredentialFormSchema], credentials: dict9 ) -> dict:10 need_validate_credential_form_schema_map = {}11 for credential_form_schema in credential_form_schemas:12 if not credential_form_schema.show_on:13 need_validate_credential_form_schema_map[credential_form_schema.variable] = credential_form_schema14 continue15 16 all_show_on_match = True17 for show_on_object in credential_form_schema.show_on:18 if show_on_object.variable not in credentials:19 all_show_on_match = False20 break21 22 if credentials[show_on_object.variable] != show_on_object.value:23 all_show_on_match = False24 break25 26 if all_show_on_match:27 need_validate_credential_form_schema_map[credential_form_schema.variable] = credential_form_schema28 29 # Iterate over the remaining credential_form_schemas, verify each credential_form_schema30 validated_credentials = {}31 for credential_form_schema in need_validate_credential_form_schema_map.values():32 # add the value of the credential_form_schema corresponding to it to validated_credentials33 result = self._validate_credential_form_schema(credential_form_schema, credentials)34 if result:35 validated_credentials[credential_form_schema.variable] = result36 37 return validated_credentials38 39 def _validate_credential_form_schema(40 self, credential_form_schema: CredentialFormSchema, credentials: dict41 ) -> Optional[str]:42 """43 Validate credential form schema44 45 :param credential_form_schema: credential form schema46 :param credentials: credentials47 :return: validated credential form schema value48 """49 # If the variable does not exist in credentials50 if credential_form_schema.variable not in credentials or not credentials[credential_form_schema.variable]:51 # If required is True, an exception is thrown52 if credential_form_schema.required:53 raise ValueError(f"Variable {credential_form_schema.variable} is required")54 else:55 # Get the value of default56 if credential_form_schema.default:57 # If it exists, add it to validated_credentials58 return credential_form_schema.default59 else:60 # If default does not exist, skip61 return None62 63 # Get the value corresponding to the variable from credentials64 value = credentials[credential_form_schema.variable]65 66 # If max_length=0, no validation is performed67 if credential_form_schema.max_length:68 if len(value) > credential_form_schema.max_length:69 raise ValueError(70 f"Variable {credential_form_schema.variable} length should not"71 f" greater than {credential_form_schema.max_length}"72 )73 74 # check the type of value75 if not isinstance(value, str):76 raise ValueError(f"Variable {credential_form_schema.variable} should be string")77 78 if credential_form_schema.type in {FormType.SELECT, FormType.RADIO}:79 # If the value is in options, no validation is performed80 if credential_form_schema.options:81 if value not in [option.value for option in credential_form_schema.options]:82 raise ValueError(f"Variable {credential_form_schema.variable} is not in options")83 84 if credential_form_schema.type == FormType.SWITCH:85 # If the value is not in ['true', 'false'], an exception is thrown86 if value.lower() not in {"true", "false"}:87 raise ValueError(f"Variable {credential_form_schema.variable} should be true or false")88 89 value = True if value.lower() == "true" else False90 91 return value92 