from ApiLayers.AllConfigs.Redis.configs import ( RedisValidationKeysAction, RedisValidationKeys, ) from ApiLayers.AllConfigs.main import LanguageConfig from Events.Engine.set_defaults.category_cluster_models import CategoryClusterController from Services.Redis.Actions.actions import RedisActions class SetDefaultLanguageModelsRedis: std_out: str = "" def __init__( self, set_response_languages_list: list[dict], set_errors_languages_list: list[dict], ): self.responses_list: list[dict] = set_response_languages_list self.errors_list: list[dict] = set_errors_languages_list def __str__(self): return f"\nPrepareLanguageModels:\n\n{self.std_out}" def set_all(self): # RedisActions.delete(list_keys=["*"]) RedisActions.delete(list_keys=[f"{RedisValidationKeys.LANGUAGE_MODELS}:*"]) for response in self.responses_list: for lang in list(LanguageConfig.SUPPORTED_LANGUAGES): for code, dict_to_set in response.items(): # [SAVE]REDIS => LANGUAGE_MODELS:STATIC:RESPONSES:{ResponseCode}:tr = {...} set_key = ( f"{RedisValidationKeysAction.static_response_key}:{code}:{lang}" ) RedisActions.set_json(list_keys=[set_key], value=dict_to_set[lang]) self.std_out += f"Language Response Models are set to Redis\n" for response in self.errors_list: for lang in list(LanguageConfig.SUPPORTED_LANGUAGES): for code, dict_to_set in response.items(): # [SAVE]REDIS => LANGUAGE_MODELS:STATIC:ERRORCODES:{ErrorCode}:en = {...} set_key = f"{RedisValidationKeysAction.static_error_code_key}:{code}:{lang}" RedisActions.set_json(list_keys=[set_key], value=dict_to_set[lang]) self.std_out += f"Language Error Models are set to Redis\n" class SetClusterLanguageModelsRedis: std_out: str = "" events_lm_dict: dict[str, dict[str, dict]] = {} events_rq_dict: dict[str, dict[str, dict]] = {} events_rs_dict: dict[str, dict[str, dict]] = {} def __init__(self, cluster_controller_group: CategoryClusterController): self.cluster_controller_group = cluster_controller_group def __str__(self): return f"\nPrepareLanguageModels:\n\n{self.std_out}" @staticmethod def merge_language_dicts(list_of_lang_models: list[dict]): """ Merges the language models of the events to a single dictionary. """ merged_lang_models: dict[str, dict] = {} for lang_model in list_of_lang_models: for lang in list(LanguageConfig.SUPPORTED_LANGUAGES): if not lang_model.get(lang, None): raise ValueError( f"Language model for {lang} not found in {lang_model}" ) if lang not in merged_lang_models: merged_lang_models[lang] = lang_model[lang] else: merged_lang_models[lang].update(lang_model[lang]) return merged_lang_models def set_models_from_cluster(self): """ iterate(ClusterToMethod) to set all models by pairing function codes """ for cluster_control in self.cluster_controller_group.imports: self.std_out += f"Setting models from cluster : {cluster_control.name}\n" for endpoint in cluster_control.category_cluster.ENDPOINTS.values(): for key_event, event in endpoint.EVENTS.items(): merged_language_dict = self.merge_language_dicts( event.LANGUAGE_MODELS ) request_validation = getattr( event.REQUEST_VALIDATOR, "model_fields", None ) response_validation = getattr( event.RESPONSE_VALIDATOR, "model_fields", None ) objects_missing = bool(request_validation) and bool( merged_language_dict ) if not objects_missing: continue if merged_language_dict: self.events_lm_dict[key_event] = merged_language_dict if request_validation: self.events_rq_dict[key_event] = request_validation if response_validation: self.events_rs_dict[key_event] = response_validation self.std_out += f"Request/Response/Language validation model is set {key_event}\n" def set_all(self): """ Set all language models from cluster list by pairing event code and models """ self.set_models_from_cluster() if self.events_lm_dict and self.events_rq_dict: """ [SAVE]REDIS => LANGUAGE_MODELS:DYNAMIC:HEADERS:REQUEST:{FunctionCode}:tr = {...} Get Request BaseModel pydantic model_fields of each event and set headers which are included in model_fields """ for lang in list( LanguageConfig.SUPPORTED_LANGUAGES ): # Iterate(languages ["tr", "en"]) for key_field in self.events_rq_dict.keys(): # Iterate(function_code) request_model = self.events_rq_dict[key_field] if not request_model: self.std_out += ( f"Request validation model not found for {key_field}\n" ) continue if ( key_field not in self.events_rq_dict or key_field not in self.events_lm_dict ): self.std_out += ( f"Request language model are missing {key_field}\n" ) continue value_to_set = {} set_key = f"{RedisValidationKeysAction.dynamic_header_request_key}:{key_field}:{lang}" for key in request_model.keys(): value_to_set[key] = self.events_lm_dict[key_field][lang][key] RedisActions.set_json(list_keys=[set_key], value=value_to_set) self.std_out += f"Language Request Headers are set to Redis\n" if self.events_lm_dict and self.events_rs_dict: """ [SAVE]REDIS => LANGUAGE_MODELS:DYNAMIC:HEADERS:RESPONSE:{FunctionCode}:en = {...} Get Response BaseModel pydantic model_fields of each event and set headers which are included in model_fields """ for lang in list( LanguageConfig.SUPPORTED_LANGUAGES ): # Iterate(languages ["tr", "en"]) for key_field in self.events_rs_dict.keys(): # Iterate(function_code) response_model = self.events_rs_dict[key_field] if not response_model: self.std_out += ( f"Response validation model not found for {key_field}\n" ) continue if ( key_field not in self.events_rs_dict or key_field not in self.events_lm_dict ): self.std_out += ( f"Response language model are missing {key_field}\n" ) continue value_to_set = {} set_key = f"{RedisValidationKeysAction.dynamic_header_response_key}:{key_field}:{lang}" for key in response_model.keys(): value_to_set[key] = self.events_lm_dict[key_field][lang][key] RedisActions.set_json(list_keys=[set_key], value=value_to_set) self.std_out += f"Language Response Headers are set to Redis\n"