import functools, hashlib, logging, operator from enum import Enum from typing import Dict, List 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 context.proto.context_pb2 import ConfigActionEnum from context.service.database.Tools import grpc_to_enum 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) def remove_dict_key(dictionary : Dict, key : str): dictionary.pop(key, None) return dictionary def config_key_hasher(resource_key : str, digest_size : int = 8): hasher = hashlib.blake2b(digest_size=digest_size) hasher.update(resource_key.encode('UTF-8')) return hasher.hexdigest() class ConfigModel(Model): 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): 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, 'resource_key': self.key, 'resource_value': self.value, } if include_position: result['position'] = self.position return result