Newer
Older
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
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