Skip to content

Utils

Tool of Paladins - A set of auxiliary methods.

accumulate_error(error_message, params)

Accumulating errors to ModelName.field_name.errors .

Source code in src/ramifice/paladins/utils.py
def accumulate_error(error_message: str, params: dict[str, Any]) -> None:
    """Accumulating errors to ModelName.field_name.errors ."""
    f__core = params["field_core"]

    if not f__core.is_hide:
        f__core.errors.append(error_message)
        if not params["is_error_symptom"]:
            params["is_error_symptom"] = True
    else:
        err_msg = (
            f">>hidden field<< -> Model: `{params['full_model_name']}` > "
            + f"Field: `{f__core.name}`"
            + f" => {error_message}"
        )
        logger.critical(err_msg)
        raise PanicError(err_msg)

check_uniqueness(value, params, field_name=None, is_is_multilingual=False) async

Checking the uniqueness of the value in the collection.

Source code in src/ramifice/paladins/utils.py
async def check_uniqueness(
    value: str | int | float,
    params: dict[str, Any],
    field_name: str | None = None,
    is_is_multilingual: bool = False,
) -> bool:
    """Checking the uniqueness of the value in the collection."""
    q_filter = None

    if is_is_multilingual:
        lang_filter = [{f"{field_name}.{params['LANG_CODE']}": value} for lang in params["LANGUAGES"]]
        q_filter = {
            "$and": [
                {"_id": {"$ne": params["doc_id"]}},
                {"$or": lang_filter},
            ],
        }
    else:
        q_filter = {
            "$and": [
                {"_id": {"$ne": params["doc_id"]}},
                {field_name: value},
            ],
        }
    return await params["collection"].find_one(q_filter) is None

ignored_fields_to_none(instance_model)

Reset the values of ignored fields to None.

Source code in src/ramifice/paladins/utils.py
def ignored_fields_to_none(instance_model: Any) -> None:
    """Reset the values of ignored fields to None."""
    descriptor_fields = instance_model.__class__.META["all_descriptor_fields"]

    for f_name in descriptor_fields:
        f__core = getattr(instance_model, f"{f_name}__core")
        if f__core.is_ignore:
            f__core.value = None
            setattr(instance_model, f_name, None)

refresh_from_mongo_doc(instance_model, mongo_doc)

Update object instance from Mongo document.

Source code in src/ramifice/paladins/utils.py
def refresh_from_mongo_doc(instance_model: Any, mongo_doc: dict[str, Any]) -> None:
    """Update object instance from Mongo document."""
    lang_code = instance_model._LANG_CODE

    for mongo_f_name, mongo_value in mongo_doc.items():
        f_name = mongo_f_name if mongo_f_name != "_id" else "id"
        field_type = getattr(instance_model, f"{f_name}__core").field_type
        f_value = None

        if mongo_value is not None:
            if field_type == "TextField":
                f_value = mongo_value.get(lang_code, "- -") if isinstance(mongo_value, dict) else mongo_value
            elif field_type == "DateField":
                f_value = mongo_value.date()
            elif field_type == "PasswordField":
                f_value = None
            else:
                f_value = mongo_value

        setattr(instance_model, f_name, f_value)