Newer
Older
# 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.
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
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
import functools, logging, operator
from enum import Enum
from typing import Dict, List, 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 device.proto.context_pb2 import ConfigActionEnum
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=False, allow_empty=True)
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
def delete_all_config_rules(database : Database, db_parent_pk : str, config_name : str) -> None:
str_config_key = key_to_str([db_parent_pk, config_name], separator=':')
db_config : ConfigModel = get_object(database, ConfigModel, str_config_key, raise_if_not_found=False)
if db_config is None: return
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)
return action, grpc_config_rule.resource_key, grpc_config_rule.resource_value
return [translate(grpc_config_rule) for grpc_config_rule in grpc_config_rules]
def get_config_rules(
database : Database, db_parent_pk : str, config_name : str
) -> List[Tuple[ORM_ConfigActionEnum, str, str]]:
str_config_key = key_to_str([db_parent_pk, config_name], separator=':')
db_config = get_object(database, ConfigModel, str_config_key, raise_if_not_found=False)
return [] if db_config is None else [
(ORM_ConfigActionEnum._value2member_map_.get(config_rule['action']),
config_rule['resource_key'], config_rule['resource_value'])
for config_rule in db_config.dump()
]
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):
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': action, 'key': resource_key,
'value': resource_value})
db_config_rule, updated = result
db_objects.append((db_config_rule, updated))
return db_objects