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
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
113
114
115
116
117
118
119
120
121
122
123
124
125
# 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 context.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=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
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)
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 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