Skip to content

Modules

OperationLogEntry

Bases: models.Model

Source code in drf_operation_log/models.py 
 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
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
class OperationLogEntry(models.Model):
    action_time = models.DateTimeField(
        _("操作时间"),
        default=timezone.now,
        db_index=True,
        editable=False,
    )
    user = models.ForeignKey(
        settings.AUTH_USER_MODEL,
        models.CASCADE,
        verbose_name=_("用户"),
    )
    content_type = models.ForeignKey(
        ContentType,
        models.SET_NULL,
        verbose_name=_("操作对象"),
        related_name="+",
        blank=True,
        null=True,
    )
    object_id = models.TextField(_("对象ID"), blank=True, null=True)
    object_repr = models.CharField(_("操作对象"), max_length=128)
    domain_content_type = models.ForeignKey(
        ContentType,
        models.SET_NULL,
        verbose_name=_("同范围操作对象"),
        related_name="+",
        blank=True,
        null=True,
    )
    domain_object_id = models.TextField(_("同范围对象ID"), blank=True, null=True)
    action = models.CharField(_("动作"), max_length=32)
    action_name = models.CharField(_("动作名称"), max_length=32)
    action_flag = models.PositiveSmallIntegerField(
        _("操作类型"), choices=ACTION_FLAG_CHOICES
    )
    change_message = models.JSONField(
        _("差异信息"), blank=True, default=dict, encoder=JSONEncoder
    )
    extra = models.JSONField(_("其他信息"), blank=True, null=True)

    objects = OperationLogEntryManager()

    class Meta:
        verbose_name = verbose_name_plural = _("操作日志")
        db_table = "drf_operation_log"
        ordering = ["-action_time"]
        index_together = ("content_type", "object_id")

    def __repr__(self):
        return str(self.action_time)

    def __str__(self):
        if self.is_addition():
            return gettext("新增 “%(object)s”.") % {"object": self.object_repr}
        elif self.is_change():
            return gettext("修改 “%(object)s” — %(changes)s") % {
                "object": self.object_repr,
                "changes": self.get_change_message(),
            }
        elif self.is_deletion():
            return gettext("删除 “%(object)s.”") % {"object": self.object_repr}

        return gettext("LogEntry Object")

    def is_addition(self):
        return self.action_flag == ADDITION

    def is_change(self):
        return self.action_flag == CHANGE

    def is_deletion(self):
        return self.action_flag == DELETION

    def get_change_message(self):
        """
        If self.change_message is a JSON structure, interpret it as a change
        string, properly translated.
        """
        # if self.is_addition():
        #     return [f"新增 {self.object_repr}"]
        # elif self.is_deletion():
        #     return [f"删除 {self.object_repr}"]
        # elif not self.change_message:
        #     return ["未更改"]

        # ret = []
        # for field_name, (old_value, new_value) in self.change_message.items():
        #     old_value = "" if old_value is None else old_value
        #     new_value = "" if new_value is None else new_value
        #     message = f"修改 {field_name}， 旧值“{old_value}”，新值“{new_value}”"
        #     ret.append(message)

        return self.change_message

    def get_edited_object(self):
        """Return the edited object represented by this log entry."""
        return self.content_type.get_object_for_this_type(pk=self.object_id)

    def get_operation_url(self):
        """
        Return the admin URL to edit the object represented by this log entry.
        """
        if self.content_type and self.object_id:
            url_name = "operationlogs-list"
            try:
                url = reverse(url_name)
                return (
                    f"{url}?content_type={self.content_type.pk}"
                    "&object_id={self.object_id}"
                )
            except NoReverseMatch:
                pass
        return None

get_change_message()

If self.change_message is a JSON structure, interpret it as a change string, properly translated.

Source code in drf_operation_log/models.py 
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
def get_change_message(self):
    """
    If self.change_message is a JSON structure, interpret it as a change
    string, properly translated.
    """
    # if self.is_addition():
    #     return [f"新增 {self.object_repr}"]
    # elif self.is_deletion():
    #     return [f"删除 {self.object_repr}"]
    # elif not self.change_message:
    #     return ["未更改"]

    # ret = []
    # for field_name, (old_value, new_value) in self.change_message.items():
    #     old_value = "" if old_value is None else old_value
    #     new_value = "" if new_value is None else new_value
    #     message = f"修改 {field_name}， 旧值“{old_value}”，新值“{new_value}”"
    #     ret.append(message)

    return self.change_message

get_edited_object()

Return the edited object represented by this log entry.

Source code in drf_operation_log/models.py 
146
147
148
def get_edited_object(self):
    """Return the edited object represented by this log entry."""
    return self.content_type.get_object_for_this_type(pk=self.object_id)

get_operation_url()

Return the admin URL to edit the object represented by this log entry.

Source code in drf_operation_log/models.py 
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
def get_operation_url(self):
    """
    Return the admin URL to edit the object represented by this log entry.
    """
    if self.content_type and self.object_id:
        url_name = "operationlogs-list"
        try:
            url = reverse(url_name)
            return (
                f"{url}?content_type={self.content_type.pk}"
                "&object_id={self.object_id}"
            )
        except NoReverseMatch:
            pass
    return None

OperationLogMixin

DRF 操作日志Mixin 必须同时满足如下三个要求的操作才会记录 1. 写操作(POST, PUT, PATCH, DELETE) 2. 未被包含在operationlog_action_exclude中的action 3. action 为 create, update, partial_update 或 destroy

自定义action需要自己创建operation_log

Source code in drf_operation_log/mixins.py 
 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
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
class OperationLogMixin:
    """
    DRF 操作日志Mixin
    必须同时满足如下三个要求的操作才会记录
    1. 写操作(POST, PUT, PATCH, DELETE)
    2. 未被包含在`operationlog_action_exclude`中的action
    3. action 为 create, update, partial_update 或 destroy

    自定义action需要自己创建operation_log
    """

    operationlog_action_exclude = []
    operationlog_domain_field: str = None

    def initial(self, request, *args, **kwargs):
        self.operation_logs = []  # noqa
        super().initial(request, *args, **kwargs)  # noqa

    def _get_view_method(self, request):
        """Get view method."""
        if hasattr(self, "action"):
            return self.action or None
        return request.method.lower()

    @staticmethod
    def _get_user(request):
        """Get user."""
        user = request.user
        if user.is_anonymous:
            return None
        return user

    def _get_action_name(self, serializer) -> str:
        """
        获取动作名称
        """
        if self.action == "create":  # noqa
            return "新增"
        elif self.action in ("update", "partial_update"):  # noqa
            return "编辑"
        elif self.action == "destroy":  # noqa
            return "删除"

        if (
            self._get_action_flag(self.request) == CHANGE
            and "action" in serializer.fields
        ):
            if isinstance(serializer.fields["action"], serializers.ChoiceField):
                action_choices = serializer.fields["action"].choices
                return action_choices.get(serializer.validated_data["action"])

        return getattr(self, self.action).kwargs["name"]  # noqa

    @staticmethod
    def _get_action_flag(request) -> int:
        if request.method == "POST":
            return ADDITION
        elif request.method in ("PUT", "PATCH"):
            return CHANGE
        elif request.method == "DELETE":
            return DELETION

    def perform_create(self, serializer):
        super().perform_create(serializer)  # noqa

        request = self.request  # noqa
        if self.should_log(request):
            operation_log = self._initial_log(
                request,
                serializer.instance,
                new_message=flatten_dict(serializer.validated_data),
            )
            self.operation_logs.append(operation_log)

    def perform_update(self, serializer):
        request = self.request  # noqa
        if self.should_log(request):
            change_message = serializer_data_diff(serializer)

            operation_log = self._initial_log(
                request,
                serializer.instance,
                change_message=change_message,
                serializer=serializer,
            )
            self.operation_logs.append(operation_log)

        super().perform_update(serializer)  # noqa

    def perform_destroy(self, instance):
        request = self.request  # noqa
        if self.should_log(request):
            operation_log = self._initial_log(request, instance)
            self.operation_logs.append(operation_log)

        super().perform_destroy(instance)  # noqa

    def get_excluded_log_fields(self, request) -> list:
        """
        获取日志排除字段
        主表字段 key1, key2
        ForeignKey key3.key31, key4.key41
        OneToMany key5[].key51, key5[].key51
        :param request:
        :return: ["key1", "key2",
            "key3.key31", "key4.key41",
            "key5[].key51", "key5[].key52"]
        """
        return []

    @action(
        detail=True,
        name="操作日志",
        serializer_class=OperationLogEntrySerializer,
    )
    def operationlogs(self, request, pk):
        """
        获取该资源操作日志接口
        :param request:
        :param pk: 主键
        :return:
        """
        excluded_fields = self.get_excluded_log_fields(request)
        queryset = OperationLogEntry.objects.select_related(
            "user", "content_type", "domain_content_type"
        ).filter(
            domain_object_id=pk,
            domain_content_type=ContentType.objects.get_for_model(self.queryset.model),
        )  # noqa

        queryset = self.filter_queryset(queryset)  # noqa
        page = self.paginate_queryset(queryset)  # noqa
        if page is not None:
            serializer = self.get_serializer(page, many=True)  # noqa
            return self.get_paginated_response(serializer.data)  # noqa

        if excluded_fields:
            formatted_excludes_fields = format_excluded_fields(excluded_fields)
            for q in queryset:
                q.changed_message = clean_excluded_fields(
                    q.changed_message,
                    formatted_excludes_fields[0],
                    formatted_excludes_fields[1],
                )

        serializer = self.get_serializer(queryset, many=True)  # noqa
        return Response(serializer.data)

    def should_log(self, request) -> bool:
        """
        是否记录操作日志, 可以覆盖此方法来自定义控制
        :param request:
        :return:
        """
        return (
            request.method.upper() in ("POST", "PUT", "PATCH", "DELETE")
            and self.action not in self.operationlog_action_exclude  # noqa
        )

    def _initial_log(
        self,
        request,
        instance,
        old_message=None,
        new_message=None,
        change_message=None,
        serializer=None,
    ) -> OperationLogEntry:
        if change_message is None and old_message and new_message and serializer:
            change_message = [
                serializer_changed_data_diff(old_message, new_message, serializer)
            ]

        action_flag = self._get_action_flag(request)

        if change_message:
            change_message = clean_data(change_message)
        else:
            if action_flag == ADDITION:
                change_message = [{"added": []}]
            elif action_flag == DELETION:
                change_message = [{"deleted": []}]

        operation_log = OperationLogEntry(
            user=request.user,
            action=self.action,  # noqa
            action_name=self._get_action_name(serializer),
            action_flag=action_flag,
            content_type=ContentType.objects.get_for_model(instance),
            object_id=instance.pk,
            object_repr=ContentType.objects.get_for_model(instance).name,
            domain_content_type=ContentType.objects.get_for_model(instance),
            domain_object_id=instance.pk,
            change_message=change_message or [],
        )

        if self.operationlog_domain_field:
            attrs = self.operationlog_domain_field.split("__")
            obj = instance
            for attr in attrs:
                obj = getattr(obj, attr)

            if not isinstance(obj, Model):
                raise ValueError("'operationlog_domain_field' must refer to a model!")

            operation_log.domain_content_type = ContentType.objects.get_for_model(obj)
            operation_log.domain_object_id = obj.pk
            operation_log.object_repr = (ContentType.objects.get_for_model(obj).name,)

        return operation_log

    def finalize_response(self, request, response, *args, **kwargs):
        if hasattr(self, "operation_logs") and not getattr(response, "exception", False):
            operation_logs_pre_save.send(
                sender="operation_logs_pre_save",
                request=request,
                operation_logs=self.operation_logs,
            )
            if getattr(settings, "DRF_OPERATION_LOG_SAVE_DATABASE", True):
                OperationLogEntry.objects.bulk_create(self.operation_logs)
            self.operation_logs.clear()

        return super().finalize_response(request, response, *args, **kwargs)  # noqa

get_excluded_log_fields(request)

获取日志排除字段 主表字段 key1, key2 ForeignKey key3.key31, key4.key41 OneToMany key5[].key51, key5[].key51 :param request: :return: ["key1", "key2", "key3.key31", "key4.key41", "key5[].key51", "key5[].key52"]

Source code in drf_operation_log/mixins.py 
126
127
128
129
130
131
132
133
134
135
136
137
def get_excluded_log_fields(self, request) -> list:
    """
    获取日志排除字段
    主表字段 key1, key2
    ForeignKey key3.key31, key4.key41
    OneToMany key5[].key51, key5[].key51
    :param request:
    :return: ["key1", "key2",
        "key3.key31", "key4.key41",
        "key5[].key51", "key5[].key52"]
    """
    return []

operationlogs(request, pk)

获取该资源操作日志接口 :param request: :param pk: 主键 :return:

Source code in drf_operation_log/mixins.py 
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
@action(
    detail=True,
    name="操作日志",
    serializer_class=OperationLogEntrySerializer,
)
def operationlogs(self, request, pk):
    """
    获取该资源操作日志接口
    :param request:
    :param pk: 主键
    :return:
    """
    excluded_fields = self.get_excluded_log_fields(request)
    queryset = OperationLogEntry.objects.select_related(
        "user", "content_type", "domain_content_type"
    ).filter(
        domain_object_id=pk,
        domain_content_type=ContentType.objects.get_for_model(self.queryset.model),
    )  # noqa

    queryset = self.filter_queryset(queryset)  # noqa
    page = self.paginate_queryset(queryset)  # noqa
    if page is not None:
        serializer = self.get_serializer(page, many=True)  # noqa
        return self.get_paginated_response(serializer.data)  # noqa

    if excluded_fields:
        formatted_excludes_fields = format_excluded_fields(excluded_fields)
        for q in queryset:
            q.changed_message = clean_excluded_fields(
                q.changed_message,
                formatted_excludes_fields[0],
                formatted_excludes_fields[1],
            )

    serializer = self.get_serializer(queryset, many=True)  # noqa
    return Response(serializer.data)

should_log(request)

是否记录操作日志, 可以覆盖此方法来自定义控制 :param request: :return:

Source code in drf_operation_log/mixins.py 
177
178
179
180
181
182
183
184
185
186
def should_log(self, request) -> bool:
    """
    是否记录操作日志, 可以覆盖此方法来自定义控制
    :param request:
    :return:
    """
    return (
        request.method.upper() in ("POST", "PUT", "PATCH", "DELETE")
        and self.action not in self.operationlog_action_exclude  # noqa
    )