Skip to content
Snippets Groups Projects
ConfigModel.py 5.79 KiB
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.

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 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=False, allow_empty=True)

    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 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)
        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 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 [
        # pylint: disable=no-member, protected-access
        (ORM_ConfigActionEnum._value2member_map_.get(config_rule['action']),
            config_rule['custom']['resource_key'], config_rule['custom']['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,
        db_config_rule, updated = result
        db_objects.append((db_config_rule, updated))

    return db_objects