Skip to content

Unit Manager

Units Management.

Management for choices parameter in dynamic field types.

UnitMixin

Units Management.

Management for choices parameter in dynamic field types.

Source code in src\ramifice\commons\unit_manager.py 
 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
class UnitMixin:
    """Units Management.

    Management for `choices` parameter in dynamic field types.
    """

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

        Management for `choices` parameter in dynamic field types.
        """
        # Get access to super collection.
        # (Contains Model state and dynamic field data.)
        super_collection: AsyncCollection = constants.MONGO_DATABASE[constants.SUPER_COLLECTION_NAME]
        # Get Model state.
        model_state: dict[str, Any] | None = await super_collection.find_one(
            filter={"collection_name": cls.META["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 = translations.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 = translations.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 = translations.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.
        cls.META["data_dynamic_fields"][unit_field] = choices or None
        # Update documents in the collection of the current Model.
        if unit.is_delete:
            collection: AsyncCollection = constants.MONGO_DATABASE[cls.META["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.

Management for choices parameter in dynamic field types.

Source code in src\ramifice\commons\unit_manager.py 
 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
@classmethod
async def unit_manager(cls: Any, unit: Unit) -> None:
    """Units Management.

    Management for `choices` parameter in dynamic field types.
    """
    # Get access to super collection.
    # (Contains Model state and dynamic field data.)
    super_collection: AsyncCollection = constants.MONGO_DATABASE[constants.SUPER_COLLECTION_NAME]
    # Get Model state.
    model_state: dict[str, Any] | None = await super_collection.find_one(
        filter={"collection_name": cls.META["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 = translations.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 = translations.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 = translations.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.
    cls.META["data_dynamic_fields"][unit_field] = choices or None
    # Update documents in the collection of the current Model.
    if unit.is_delete:
        collection: AsyncCollection = constants.MONGO_DATABASE[cls.META["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,
            )