Skip to content

Save

Create or update document in database.

SaveMixin

Create or update document in database.

Source code in src/ramifice/paladins/save.py 
 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
class SaveMixin:
    """Create or update document in database."""

    async def save(self) -> bool:
        """Create or update document in database.

        This method pre-uses the `check` method.
        """
        metadata = self.__class__.META
        # Get collection.
        collection: AsyncCollection = Config.MONGO_DATABASE[metadata["collection_name"]]
        # Check Model.
        result_check: dict[str, Any] = await self.check(is_save=True, collection=collection)
        # Reset the alerts to exclude duplicates.
        self.id__core.alerts = []
        # Check the conditions and, if necessary, define a message for the web form.
        if not result_check["is_update"] and not metadata["is_create_doc"]:
            self.id__core.alerts.append("It is forbidden to create new documents !")
            result_check["is_valid"] = False
        if result_check["is_update"] and not metadata["is_update_doc"]:
            self.id__core.alerts.append("It is forbidden to update documents !")
            result_check["is_valid"] = False
        # Leave the method if the check fails.
        if not result_check["is_valid"]:
            ignored_fields_to_none(self)
            return False
        # Get data for document.
        checked_data: dict[str, Any] = result_check["data"]
        # Create or update a document in database.
        if result_check["is_update"]:
            # Update date and time.
            checked_data["updated_at"] = datetime.now(Config.UTC_TIMEZONE)
            # Run hook.
            await self.pre_update()
            # Update doc.
            await collection.update_one({"_id": checked_data["_id"]}, {"$set": checked_data})
            # Run hook.
            await self.post_update()
            # Refresh Model.
            mongo_doc: dict[str, Any] | None = await collection.find_one({"_id": checked_data["_id"]})
            if mongo_doc is None:
                err_msg = (
                    f"Model: `{self.full_model_name()}` > "
                    + "Method: `save` => "
                    + "Geted value is None - it is impossible to refresh the current Model."
                )
                logger.critical(err_msg)
                raise PanicError(err_msg)
            refresh_from_mongo_doc(self, mongo_doc)
        else:
            # Add date and time.
            today = datetime.now(Config.UTC_TIMEZONE)
            checked_data["created_at"] = today
            checked_data["updated_at"] = today
            # Run hook.
            await self.pre_create()
            # Insert doc.
            await collection.insert_one(checked_data)
            # Run hook.
            await self.post_create()
            # Refresh Model.
            mongo_doc = await collection.find_one({"_id": checked_data["_id"]})
            if mongo_doc is None:
                err_msg = (
                    f"Model: `{self.full_model_name()}` > " + "Method: `save` => " + "The document was not created."
                )
                logger.critical(err_msg)
                raise PanicError(err_msg)
            refresh_from_mongo_doc(self, mongo_doc)
        #
        # If everything is completed successfully.
        return True

save() async

Create or update document in database.

This method pre-uses the check method.

Source code in src/ramifice/paladins/save.py 
 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
async def save(self) -> bool:
    """Create or update document in database.

    This method pre-uses the `check` method.
    """
    metadata = self.__class__.META
    # Get collection.
    collection: AsyncCollection = Config.MONGO_DATABASE[metadata["collection_name"]]
    # Check Model.
    result_check: dict[str, Any] = await self.check(is_save=True, collection=collection)
    # Reset the alerts to exclude duplicates.
    self.id__core.alerts = []
    # Check the conditions and, if necessary, define a message for the web form.
    if not result_check["is_update"] and not metadata["is_create_doc"]:
        self.id__core.alerts.append("It is forbidden to create new documents !")
        result_check["is_valid"] = False
    if result_check["is_update"] and not metadata["is_update_doc"]:
        self.id__core.alerts.append("It is forbidden to update documents !")
        result_check["is_valid"] = False
    # Leave the method if the check fails.
    if not result_check["is_valid"]:
        ignored_fields_to_none(self)
        return False
    # Get data for document.
    checked_data: dict[str, Any] = result_check["data"]
    # Create or update a document in database.
    if result_check["is_update"]:
        # Update date and time.
        checked_data["updated_at"] = datetime.now(Config.UTC_TIMEZONE)
        # Run hook.
        await self.pre_update()
        # Update doc.
        await collection.update_one({"_id": checked_data["_id"]}, {"$set": checked_data})
        # Run hook.
        await self.post_update()
        # Refresh Model.
        mongo_doc: dict[str, Any] | None = await collection.find_one({"_id": checked_data["_id"]})
        if mongo_doc is None:
            err_msg = (
                f"Model: `{self.full_model_name()}` > "
                + "Method: `save` => "
                + "Geted value is None - it is impossible to refresh the current Model."
            )
            logger.critical(err_msg)
            raise PanicError(err_msg)
        refresh_from_mongo_doc(self, mongo_doc)
    else:
        # Add date and time.
        today = datetime.now(Config.UTC_TIMEZONE)
        checked_data["created_at"] = today
        checked_data["updated_at"] = today
        # Run hook.
        await self.pre_create()
        # Insert doc.
        await collection.insert_one(checked_data)
        # Run hook.
        await self.post_create()
        # Refresh Model.
        mongo_doc = await collection.find_one({"_id": checked_data["_id"]})
        if mongo_doc is None:
            err_msg = (
                f"Model: `{self.full_model_name()}` > " + "Method: `save` => " + "The document was not created."
            )
            logger.critical(err_msg)
            raise PanicError(err_msg)
        refresh_from_mongo_doc(self, mongo_doc)
    #
    # If everything is completed successfully.
    return True