Skip to content

Unit Manager

Units Management.

Management for choices parameter in dynamic fields.

UnitMixin

Units Management.

Source code in src/ramifice/commons/unit_manager.py 
 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
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
class UnitMixin:
    """Units Management."""

    @classmethod
    async def unit_manager(cls: Any, unit: Unit) -> None:
        """Units Management.

        Unit - A unit of information ([value, "Title"]) for fields with a dynamic `choices` attribute.
        Management for `choices` parameter in dynamic fields.
        """
        metadata = cls.META
        # Get access to super collection.
        # (Contains Model state and dynamic field data.)
        super_collection: AsyncCollection = Config.MONGO_DATABASE[Config.SUPER_COLLECTION_NAME]
        # Get Model state.
        model_state: dict[str, Any] | None = await super_collection.find_one(
            filter={"collection_name": metadata["collection_name"]},
        )
        # Check the presence of a Model state.
        if model_state is None:
            msg = "Error: Model State - Not found!"
            logger.critical(msg)
            raise PanicError(msg)
        # Get language list.
        lang_list = deepcopy(Translator.LANGUAGES)
        # Get clean fields of Unit.
        unit_field: str = unit.field
        title = unit.title
        if len(title) != len(lang_list):
            msg = "Unit.title => There are no translations for some languages!"
            logger.critical(msg)
            raise PanicError(msg)
        title = {lang: title[lang] for lang in lang_list}
        target_value = unit.value
        # Get dynamic field data.
        choices: list | None = model_state["data_dynamic_fields"][unit_field]
        # Determine the presence of unit.
        is_unit_exists: bool = False
        if choices is not None:
            for item in choices:
                if item["value"] == target_value:
                    is_unit_exists = True
                    break
        # Add Unit to Model State.
        if not unit.is_delete:
            if choices is not None:
                if is_unit_exists:
                    main_lang = Translator.DEFAULT_LOCALE
                    msg = (
                        "Error: It is not possible to add Unit - "
                        + f"Unit `{title[main_lang]}: {target_value}` is exists!"
                    )
                    logger.error(msg)
                    raise NotPossibleAddUnitError(msg)
                choices.append({"title": title, "value": target_value})
            else:
                choices = [{"title": title, "value": target_value}]
            model_state["data_dynamic_fields"][unit_field] = choices
        else:
            # Delete Unit from Model State.
            if choices is None:
                msg = "Error: It is not possible to delete Unit - Units is not exists!"
                logger.error(msg)
                raise NotPossibleDeleteUnitError(msg)
            if not is_unit_exists:
                main_lang = Translator.DEFAULT_LOCALE
                msg = (
                    "Error: It is not possible to delete Unit."
                    + f"Unit `{title[main_lang]}: {target_value}` is not exists!"
                )
                logger.erro(msg)
                raise NotPossibleDeleteUnitError(msg)
            choices = [item for item in choices if item["value"] != target_value]
            model_state["data_dynamic_fields"][unit_field] = choices or None
        # Update state of current Model in super collection.
        await super_collection.replace_one(
            filter={"collection_name": model_state["collection_name"]},
            replacement=model_state,
        )
        # Update metadata of current Model.
        metadata["data_dynamic_fields"][unit_field] = choices or None
        # Update documents in the collection of the current Model.
        if unit.is_delete:
            collection: AsyncCollection = Config.MONGO_DATABASE[metadata["collection_name"]]
            async for mongo_doc in collection.find():
                field_value = mongo_doc[unit_field]
                if field_value is not None:
                    if isinstance(field_value, list):
                        value_list = mongo_doc[unit_field]
                        value_list.remove(target_value)
                        mongo_doc[unit_field] = value_list or None
                    else:
                        mongo_doc[unit_field] = None
                await collection.replace_one(
                    filter={"_id": mongo_doc["_id"]},
                    replacement=mongo_doc,
                )

unit_manager(unit) async classmethod

Units Management.

Unit - A unit of information ([value, "Title"]) for fields with a dynamic choices attribute. Management for choices parameter in dynamic fields.

Source code in src/ramifice/commons/unit_manager.py 
 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
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
@classmethod
async def unit_manager(cls: Any, unit: Unit) -> None:
    """Units Management.

    Unit - A unit of information ([value, "Title"]) for fields with a dynamic `choices` attribute.
    Management for `choices` parameter in dynamic fields.
    """
    metadata = cls.META
    # Get access to super collection.
    # (Contains Model state and dynamic field data.)
    super_collection: AsyncCollection = Config.MONGO_DATABASE[Config.SUPER_COLLECTION_NAME]
    # Get Model state.
    model_state: dict[str, Any] | None = await super_collection.find_one(
        filter={"collection_name": metadata["collection_name"]},
    )
    # Check the presence of a Model state.
    if model_state is None:
        msg = "Error: Model State - Not found!"
        logger.critical(msg)
        raise PanicError(msg)
    # Get language list.
    lang_list = deepcopy(Translator.LANGUAGES)
    # Get clean fields of Unit.
    unit_field: str = unit.field
    title = unit.title
    if len(title) != len(lang_list):
        msg = "Unit.title => There are no translations for some languages!"
        logger.critical(msg)
        raise PanicError(msg)
    title = {lang: title[lang] for lang in lang_list}
    target_value = unit.value
    # Get dynamic field data.
    choices: list | None = model_state["data_dynamic_fields"][unit_field]
    # Determine the presence of unit.
    is_unit_exists: bool = False
    if choices is not None:
        for item in choices:
            if item["value"] == target_value:
                is_unit_exists = True
                break
    # Add Unit to Model State.
    if not unit.is_delete:
        if choices is not None:
            if is_unit_exists:
                main_lang = Translator.DEFAULT_LOCALE
                msg = (
                    "Error: It is not possible to add Unit - "
                    + f"Unit `{title[main_lang]}: {target_value}` is exists!"
                )
                logger.error(msg)
                raise NotPossibleAddUnitError(msg)
            choices.append({"title": title, "value": target_value})
        else:
            choices = [{"title": title, "value": target_value}]
        model_state["data_dynamic_fields"][unit_field] = choices
    else:
        # Delete Unit from Model State.
        if choices is None:
            msg = "Error: It is not possible to delete Unit - Units is not exists!"
            logger.error(msg)
            raise NotPossibleDeleteUnitError(msg)
        if not is_unit_exists:
            main_lang = Translator.DEFAULT_LOCALE
            msg = (
                "Error: It is not possible to delete Unit."
                + f"Unit `{title[main_lang]}: {target_value}` is not exists!"
            )
            logger.erro(msg)
            raise NotPossibleDeleteUnitError(msg)
        choices = [item for item in choices if item["value"] != target_value]
        model_state["data_dynamic_fields"][unit_field] = choices or None
    # Update state of current Model in super collection.
    await super_collection.replace_one(
        filter={"collection_name": model_state["collection_name"]},
        replacement=model_state,
    )
    # Update metadata of current Model.
    metadata["data_dynamic_fields"][unit_field] = choices or None
    # Update documents in the collection of the current Model.
    if unit.is_delete:
        collection: AsyncCollection = Config.MONGO_DATABASE[metadata["collection_name"]]
        async for mongo_doc in collection.find():
            field_value = mongo_doc[unit_field]
            if field_value is not None:
                if isinstance(field_value, list):
                    value_list = mongo_doc[unit_field]
                    value_list.remove(target_value)
                    mongo_doc[unit_field] = value_list or None
                else:
                    mongo_doc[unit_field] = None
            await collection.replace_one(
                filter={"_id": mongo_doc["_id"]},
                replacement=mongo_doc,
            )