69 lines
2.3 KiB
Python
69 lines
2.3 KiB
Python
from typing import Optional, Dict, List
|
|
import json
|
|
import logging
|
|
|
|
from django.contrib.admin.models import CHANGE, ACTION_FLAG_CHOICES, LogEntry
|
|
from django.contrib.contenttypes.models import ContentType
|
|
import django.conf
|
|
import django.db.models
|
|
|
|
from common.middleware import threadlocal
|
|
|
|
logger = logging.getLogger(__name__)
|
|
ACTION_FLAG_DISPLAY = {k: v for k, v in ACTION_FLAG_CHOICES}
|
|
|
|
|
|
def attach_log_entry(
|
|
instance: django.db.models.Model,
|
|
message: List[Dict[str, str]],
|
|
action_flag: int = CHANGE,
|
|
user_id: Optional[int] = None,
|
|
) -> None:
|
|
"""Attach a log entry to this model, for in the admin 'history' page.
|
|
|
|
:param instance:
|
|
:param message:
|
|
:param action_flag: Either ADDITION, CHANGE, or DELETION, defaults to CHANGE.
|
|
:param user_id: Optionally, the user who performs the logged action.
|
|
Can be None when the action is performed by the system.
|
|
"""
|
|
from django.contrib.auth import get_user_model
|
|
|
|
User = get_user_model()
|
|
|
|
if user_id is None:
|
|
request = threadlocal.get_current_request()
|
|
# N.B.: request.user.pk can be None after login of a new user
|
|
if request and request.user.is_authenticated and request.user.pk:
|
|
user_id = request.user.pk
|
|
else:
|
|
user_id = django.conf.settings.SYSTEM_USER_ID
|
|
|
|
object_repr = repr(instance)
|
|
logger.info('%s of "%s": %s', ACTION_FLAG_DISPLAY[action_flag], object_repr, message)
|
|
# Check that user referred to by the entry actually exists
|
|
if not User.objects.filter(pk=user_id).exists():
|
|
logger.error(
|
|
'Cannot record change "%s" to %s: user pk=%s does not exist',
|
|
message,
|
|
object_repr,
|
|
user_id,
|
|
)
|
|
return
|
|
|
|
LogEntry.objects.log_action(
|
|
user_id=user_id,
|
|
content_type_id=ContentType.objects.get_for_model(type(instance)).pk,
|
|
object_id=instance.pk,
|
|
object_repr=object_repr,
|
|
action_flag=action_flag,
|
|
change_message=json.dumps(message, default=str),
|
|
)
|
|
|
|
|
|
def entries_for(instance: django.db.models.Model) -> 'django.db.models.QuerySet[LogEntry]':
|
|
"""Build a query for all log entries attached to an instance."""
|
|
return LogEntry.objects.filter(
|
|
content_type_id=ContentType.objects.get_for_model(type(instance)).pk, object_id=instance.pk
|
|
)
|