Skip to content

Delete

Delete document from database.

DeleteMixin

Delete document from database.

Source code in src/ramifice/paladins/delete.py 
 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
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
class DeleteMixin:
    """Delete document from database."""

    async def delete(
        self,
        remove_files: bool = True,
        projection: Any | None = None,
        sort: Any | None = None,
        hint: Any | None = None,
        session: Any | None = None,
        let: Any | None = None,
        comment: Any | None = None,
        **kwargs: dict[str, Any],
    ) -> dict[str, Any]:
        """Delete document from database.

        Agrs:
            remove_files: True, if need to delete files and images in the `public/media/uploads` directory.
            projection: A  list of field names that should be
                            returned in the result document or a mapping specifying the fields
                            to include or exclude. If projection is a list "_id" will
                            always be returned. Use a mapping to exclude fields from
                            the result (e.g. projection={'_id': False}).
            sort: A  list of (key, direction) pairs
                    specifying the sort order for the query. If multiple documents
                    match the query, they are sorted and the first is deleted.
            hint: An index to use to support the query predicate
                    specified either by its string name, or in the same format as
                    passed to ~pymongo.asynchronous.collection.AsyncCollection.create_index
                    (e.g. [('field', ASCENDING)]). This option is only supported
                    on MongoDB 4.4 and above.
            session: A ~pymongo.asynchronous.client_session.AsyncClientSession.
            let: Map of parameter names and values. Values must be
                    constant or closed expressions that do not reference document
                    fields. Parameters can then be accessed as variables in an
                    aggregate expression context (e.g. "$$var").
            comment: A user-provided comment to attach to this command.
        """
        metadata = self.__class__.META
        # Raises a panic if the Model cannot be removed.
        if not metadata["is_delete_doc"]:
            err_msg = (
                f"Model: `{metadata['full_model_name']}` > "
                + "META param: `is_delete_doc` (False) => "
                + "Documents of this Model cannot be removed from the database!"
            )
            logger.warning(err_msg)
            raise ForbiddenDeleteDocError(err_msg)
        # Get documet ID.
        doc_id = self.id
        if doc_id is None:
            err_msg = f"Model: `{metadata['full_model_name']}` > " + "Field: `id` => ID is missing."
            logger.critical(err_msg)
            raise PanicError(err_msg)
        # Run hook.
        await self.pre_delete()
        # Get collection for current Model.
        collection: AsyncCollection = Config.MONGO_DATABASE[metadata["collection_name"]]
        # Delete document.
        mongo_doc: dict[str, Any] | None = {}
        mongo_doc = await collection.find_one_and_delete(
            filter={"_id": doc_id},
            projection=projection,
            sort=sort,
            hint=hint,
            session=session,
            let=let,
            comment=comment,
            **kwargs,
        )
        # If the document failed to delete.
        if not bool(mongo_doc):
            err_msg = (
                f"Model: `{metadata['full_model_name']}` > "
                + "Method: `delete` => "
                + "The document was not deleted, the document is absent in the database."
            )
            logger.critical(err_msg)
            raise PanicError(err_msg)
        # Delete orphaned files and add None to field.value.
        file_data: dict[str, Any] | None = None
        for f_name in metadata["all_descriptor_fields"]:
            f__core = getattr(self, f"{f_name}__core")
            if remove_files and not f__core.is_ignore:
                group = f__core.group
                if group == "file":
                    file_data = mongo_doc[f_name]
                    if file_data is not None and len(f__core.value["path"]) > 0:
                        await to_thread.run_sync(remove, f__core.value["path"])
                    file_data = None
                elif group == "img":
                    file_data = mongo_doc[f_name]
                    if file_data is not None and len(f__core.value["imgs_dir_path"]) > 0:
                        # pyrefly: ignore [incompatible-overload-residual]
                        await to_thread.run_sync(rmtree, f__core.value["imgs_dir_path"])
                    file_data = None
            setattr(self, f_name, None)
        # Run hook.
        await self.post_delete()
        #
        return mongo_doc

delete(remove_files=True, projection=None, sort=None, hint=None, session=None, let=None, comment=None, **kwargs) async

Delete document from database.

Agrs

remove_files: True, if need to delete files and images in the public/media/uploads directory. projection: A list of field names that should be returned in the result document or a mapping specifying the fields to include or exclude. If projection is a list "_id" will always be returned. Use a mapping to exclude fields from the result (e.g. projection={'_id': False}). sort: A list of (key, direction) pairs specifying the sort order for the query. If multiple documents match the query, they are sorted and the first is deleted. hint: An index to use to support the query predicate specified either by its string name, or in the same format as passed to ~pymongo.asynchronous.collection.AsyncCollection.create_index (e.g. [('field', ASCENDING)]). This option is only supported on MongoDB 4.4 and above. session: A ~pymongo.asynchronous.client_session.AsyncClientSession. let: Map of parameter names and values. Values must be constant or closed expressions that do not reference document fields. Parameters can then be accessed as variables in an aggregate expression context (e.g. "$$var"). comment: A user-provided comment to attach to this command.

Source code in src/ramifice/paladins/delete.py 
 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
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
async def delete(
    self,
    remove_files: bool = True,
    projection: Any | None = None,
    sort: Any | None = None,
    hint: Any | None = None,
    session: Any | None = None,
    let: Any | None = None,
    comment: Any | None = None,
    **kwargs: dict[str, Any],
) -> dict[str, Any]:
    """Delete document from database.

    Agrs:
        remove_files: True, if need to delete files and images in the `public/media/uploads` directory.
        projection: A  list of field names that should be
                        returned in the result document or a mapping specifying the fields
                        to include or exclude. If projection is a list "_id" will
                        always be returned. Use a mapping to exclude fields from
                        the result (e.g. projection={'_id': False}).
        sort: A  list of (key, direction) pairs
                specifying the sort order for the query. If multiple documents
                match the query, they are sorted and the first is deleted.
        hint: An index to use to support the query predicate
                specified either by its string name, or in the same format as
                passed to ~pymongo.asynchronous.collection.AsyncCollection.create_index
                (e.g. [('field', ASCENDING)]). This option is only supported
                on MongoDB 4.4 and above.
        session: A ~pymongo.asynchronous.client_session.AsyncClientSession.
        let: Map of parameter names and values. Values must be
                constant or closed expressions that do not reference document
                fields. Parameters can then be accessed as variables in an
                aggregate expression context (e.g. "$$var").
        comment: A user-provided comment to attach to this command.
    """
    metadata = self.__class__.META
    # Raises a panic if the Model cannot be removed.
    if not metadata["is_delete_doc"]:
        err_msg = (
            f"Model: `{metadata['full_model_name']}` > "
            + "META param: `is_delete_doc` (False) => "
            + "Documents of this Model cannot be removed from the database!"
        )
        logger.warning(err_msg)
        raise ForbiddenDeleteDocError(err_msg)
    # Get documet ID.
    doc_id = self.id
    if doc_id is None:
        err_msg = f"Model: `{metadata['full_model_name']}` > " + "Field: `id` => ID is missing."
        logger.critical(err_msg)
        raise PanicError(err_msg)
    # Run hook.
    await self.pre_delete()
    # Get collection for current Model.
    collection: AsyncCollection = Config.MONGO_DATABASE[metadata["collection_name"]]
    # Delete document.
    mongo_doc: dict[str, Any] | None = {}
    mongo_doc = await collection.find_one_and_delete(
        filter={"_id": doc_id},
        projection=projection,
        sort=sort,
        hint=hint,
        session=session,
        let=let,
        comment=comment,
        **kwargs,
    )
    # If the document failed to delete.
    if not bool(mongo_doc):
        err_msg = (
            f"Model: `{metadata['full_model_name']}` > "
            + "Method: `delete` => "
            + "The document was not deleted, the document is absent in the database."
        )
        logger.critical(err_msg)
        raise PanicError(err_msg)
    # Delete orphaned files and add None to field.value.
    file_data: dict[str, Any] | None = None
    for f_name in metadata["all_descriptor_fields"]:
        f__core = getattr(self, f"{f_name}__core")
        if remove_files and not f__core.is_ignore:
            group = f__core.group
            if group == "file":
                file_data = mongo_doc[f_name]
                if file_data is not None and len(f__core.value["path"]) > 0:
                    await to_thread.run_sync(remove, f__core.value["path"])
                file_data = None
            elif group == "img":
                file_data = mongo_doc[f_name]
                if file_data is not None and len(f__core.value["imgs_dir_path"]) > 0:
                    # pyrefly: ignore [incompatible-overload-residual]
                    await to_thread.run_sync(rmtree, f__core.value["imgs_dir_path"])
                file_data = None
        setattr(self, f_name, None)
    # Run hook.
    await self.post_delete()
    #
    return mongo_doc