# Copyright 2021-2023 H2020 TeraFlow (https://www.teraflow-h2020.eu/) # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. import functools, logging, operator from enum import Enum from typing import Dict, List, Optional, Tuple, Union from common.orm.Database import Database from common.orm.HighLevel import get_object, get_or_create_object, update_or_create_object from common.orm.backend.Tools import key_to_str from common.orm.fields.EnumeratedField import EnumeratedField from common.orm.fields.ForeignKeyField import ForeignKeyField from common.orm.fields.IntegerField import IntegerField from common.orm.fields.PrimaryKeyField import PrimaryKeyField from common.orm.fields.StringField import StringField from common.orm.model.Model import Model from common.proto.context_pb2 import ConfigActionEnum from common.tools.grpc.Tools import grpc_message_to_json_string from .Tools import fast_hasher, grpc_to_enum, remove_dict_key LOGGER = logging.getLogger(__name__) class ORM_ConfigActionEnum(Enum): UNDEFINED = ConfigActionEnum.CONFIGACTION_UNDEFINED SET = ConfigActionEnum.CONFIGACTION_SET DELETE = ConfigActionEnum.CONFIGACTION_DELETE grpc_to_enum__config_action = functools.partial( grpc_to_enum, ConfigActionEnum, ORM_ConfigActionEnum) class ConfigModel(Model): # pylint: disable=abstract-method pk = PrimaryKeyField() def dump(self) -> List[Dict]: db_config_rule_pks = self.references(ConfigRuleModel) config_rules = [ConfigRuleModel(self.database, pk).dump(include_position=True) for pk,_ in db_config_rule_pks] config_rules = sorted(config_rules, key=operator.itemgetter('position')) return [remove_dict_key(config_rule, 'position') for config_rule in config_rules] class ConfigRuleModel(Model): # pylint: disable=abstract-method pk = PrimaryKeyField() config_fk = ForeignKeyField(ConfigModel) position = IntegerField(min_value=0, required=True) action = EnumeratedField(ORM_ConfigActionEnum, required=True) key = StringField(required=True, allow_empty=False) value = StringField(required=True, allow_empty=False) def dump(self, include_position=True) -> Dict: # pylint: disable=arguments-differ result = { 'action': self.action.value, 'custom': { 'resource_key': self.key, 'resource_value': self.value, }, } if include_position: result['position'] = self.position return result def set_config_rule( database : Database, db_config : ConfigModel, position : int, resource_key : str, resource_value : str ) -> Tuple[ConfigRuleModel, bool]: str_rule_key_hash = fast_hasher(resource_key) str_config_rule_key = key_to_str([db_config.pk, str_rule_key_hash], separator=':') result : Tuple[ConfigRuleModel, bool] = update_or_create_object(database, ConfigRuleModel, str_config_rule_key, { 'config_fk': db_config, 'position': position, 'action': ORM_ConfigActionEnum.SET, 'key': resource_key, 'value': resource_value}) db_config_rule, updated = result return db_config_rule, updated def delete_config_rule( database : Database, db_config : ConfigModel, resource_key : str ) -> None: str_rule_key_hash = fast_hasher(resource_key) str_config_rule_key = key_to_str([db_config.pk, str_rule_key_hash], separator=':') db_config_rule : Optional[ConfigRuleModel] = get_object( database, ConfigRuleModel, str_config_rule_key, raise_if_not_found=False) if db_config_rule is None: return db_config_rule.delete() def delete_all_config_rules( database : Database, db_config : ConfigModel ) -> None: db_config_rule_pks = db_config.references(ConfigRuleModel) for pk,_ in db_config_rule_pks: ConfigRuleModel(database, pk).delete() def grpc_config_rules_to_raw(grpc_config_rules) -> List[Tuple[ORM_ConfigActionEnum, str, str]]: def translate(grpc_config_rule): action = grpc_to_enum__config_action(grpc_config_rule.action) config_rule_type = str(grpc_config_rule.WhichOneof('config_rule')) if config_rule_type != 'custom': raise NotImplementedError('ConfigRule of type {:s} is not implemented: {:s}'.format( config_rule_type, grpc_message_to_json_string(grpc_config_rule))) return action, grpc_config_rule.custom.resource_key, grpc_config_rule.custom.resource_value return [translate(grpc_config_rule) for grpc_config_rule in grpc_config_rules] def update_config( database : Database, db_parent_pk : str, config_name : str, raw_config_rules : List[Tuple[ORM_ConfigActionEnum, str, str]] ) -> List[Tuple[Union[ConfigModel, ConfigRuleModel], bool]]: str_config_key = key_to_str([db_parent_pk, config_name], separator=':') result : Tuple[ConfigModel, bool] = get_or_create_object(database, ConfigModel, str_config_key) db_config, created = result db_objects : List[Tuple[Union[ConfigModel, ConfigRuleModel], bool]] = [(db_config, created)] for position,(action, resource_key, resource_value) in enumerate(raw_config_rules): if action == ORM_ConfigActionEnum.SET: result : Tuple[ConfigRuleModel, bool] = set_config_rule( database, db_config, position, resource_key, resource_value) db_config_rule, updated = result db_objects.append((db_config_rule, updated)) elif action == ORM_ConfigActionEnum.DELETE: delete_config_rule(database, db_config, resource_key) else: msg = 'Unsupported action({:s}) for resource_key({:s})/resource_value({:s})' raise AttributeError(msg.format(str(ConfigActionEnum.Name(action)), str(resource_key), str(resource_value))) return db_objects