🧰

DRF(django-rest-framework)でAPIリクエストのアクセスログをDBに保存する(カスタムmiddleware)

に公開
Django
Python
tech

概要

django-rest-frameworkを使ったバックエンドサーバへのクライアントからのリクエストをDBに保存する方法を備忘として書く。
※django-rest-frameworkに限らず通常のdjango?でも使えるはず...

動くサンプルコードはこちら。記事中のコードは端折っている部分もあるので完全なのを見たい場合はこちらを見てください。

  • logアプリ: APIリクエストのログ記録と記録したログを表示するviewを定義してます
  • apiアプリ: 検証用のエラーを起こすviewを定義してます

https://github.com/nelsia/django-activity-log-example

この記事に書いてあること

主に以下のことを書いています。

  • djangoサーバへのリクエストをDBに記録するカスタムミドルウェアを作成
  • リクエストの結果が500エラーになった場合にはDBにdjangoのtechnical_500_responseを記録
  • 記録したリクエストログを取得するViewの作成

注意事項

以下のライブラリバージョンで動作確認しています。

Django==4.2
djangorestframework==3.15.2

作成するコード

必要になるコードだけピックアップして書いていきます。
なお、前提としてlogアプリを作成して、settings.pyのINSTALLED_APPSに追加しておいてください。

$ python manage.py startapp log

1. リクエストログを記録するモデルを作成

log/models.py
from django.db import models

class ActivityLog(models.Model):
    # Request Info
    path = models.CharField(max_length=255)
    method = models.CharField(max_length=10)
    query_params = models.TextField(null=True, blank=True)
    headers = models.TextField(null=True, blank=True)
    body = models.TextField(null=True, blank=True)

    # Response Info
    response_code = models.PositiveIntegerField()
    response_body = models.TextField(null=True, blank=True)

    # Meta Info
    ip_address = models.GenericIPAddressField(null=True, blank=True)
    user_agent = models.CharField(max_length=512, null=True, blank=True)

    # Django Traceback HTML
    traceback_html = models.TextField(null=True, blank=True)

    request_at = models.DateTimeField(auto_now_add=True)

    def __str__(self):
        return str(self.path)

2. カスタムミドルウェアの作成

カスタムミドルウェアを作成します。
例外が発生したときにはdef process_exceptionでdjangoのDEBUG=Trueの時に見れるエラー画面？のHTMLをDBに格納します。

こんな感じの画面のHTMLを格納する

log/middleware/activity_log_middleware.py
import sys

from django.views import debug

from log.models import ActivityLog


class ActivityLogMiddleware:
    def __init__(self, get_response):
        self.get_response = get_response
        self.traceback_html = None # technical 500 error html

    def __call__(self, request):
        request_body = request.body
        response = self.get_response(request)
        self.process_request(request, response, request_body)
        return response

    def process_request(self, request, response, request_body):
        # /log/から始まるパスのAPIリクエスト以外を記録
        if not request.path.startswith("/log/"):
            try:
                ActivityLog.objects.create(
                    path=request.path,
                    method=request.method,
                    query_params=request.META.get("QUERY_STRING", ""),
                    headers=str(request.headers),
                    body=request_body.decode("utf-8") if request_body else None,
                    response_body=response.content.decode("utf-8") if response.content else None,
                    response_code=response.status_code,
                    ip_address=request.META.get("HTTP_X_CLIENT_SOURCE", request.META.get("REMOTE_ADDR")),
                    user_agent=request.META.get("HTTP_USER_AGENT", None),
                    traceback_html=self.traceback_html,
                )

            except Exception as e:
                print(e)

        return None

    def process_exception(self, request, exception):
        """
        例外が発生したときにはtechnical_500_responseのHTMLをDBに格納する
        """
        exc_info = sys.exc_info()
        self.traceback_html = debug.technical_500_response(request, *exc_info).content.decode("utf-8")

3. settings.pyにミドルウェアを登録

settings.py
MIDDLEWARE = [
    'corsheaders.middleware.CorsMiddleware',
    'django.middleware.security.SecurityMiddleware',
    'django.contrib.sessions.middleware.SessionMiddleware',
    'django.middleware.common.CommonMiddleware',
    'django.middleware.csrf.CsrfViewMiddleware',
    'django.contrib.auth.middleware.AuthenticationMiddleware',
    'django.contrib.messages.middleware.MessageMiddleware',
    'django.middleware.clickjacking.XFrameOptionsMiddleware',

    'log.middleware.activity_log_middleware.ActivityLogMiddleware', # 追記
]

4. 記録したログを表示するViewとSerializerを作成

django-rest-frameworkのSerializerとAPIViewを使って、DBに記録したログを表示します。
500エラーのログで、djangoのtechnical_500_errorで取得したHTMLがDBに格納されている場合には表示するURLのリンクとしてtraceback_urlを返します。

log/serializers.py
from django.urls import reverse
from rest_framework import serializers

from .models import ActivityLog


class ActivityLogSerializer(serializers.ModelSerializer):
    traceback_url = serializers.SerializerMethodField()

    class Meta:
        model = ActivityLog
        fields = ["id", "path", "method", "query_params", "headers", "body", "response_code", 
                  "ip_address", "user_agent", "request_at", "traceback_url"]

    def get_traceback_url(self, obj):
        if obj.traceback_html is None:
            return ""

        request = self.context.get('request')
        if request:
            return request.build_absolute_uri(reverse("activity_log", kwargs={"log_id": obj.id}))
        return reverse("activity_log", kwargs={"log_id": obj.id})
log/views.py
from django.http import HttpResponse, HttpResponseNotFound
from rest_framework.pagination import LimitOffsetPagination
from rest_framework.views import APIView

from .models import ActivityLog
from .serializers import ActivityLogSerializer


class ActivityLogView(APIView, LimitOffsetPagination):
    def get(self, request):
        qs = ActivityLog.objects.all().order_by("-id")

        paginate_qs = self.paginate_queryset(qs, request, view=self)
        serializer = ActivityLogSerializer(paginate_qs, many=True, context={'request': request})

        res = {
            "data": serializer.data
        }

        return self.get_paginated_response(res)


def view_traceback_html(request, log_id):
    """
    Traceback html view
    """
    try:
        traceback_html = ActivityLog.objects.get(pk=log_id).traceback_html
    except ActivityLog.DoesNotExist:
        return HttpResponseNotFound("Activity Log Not Found")
    
    return HttpResponse(traceback_html,  content_type='text/html')

5. urls.py

urls.py
from django.urls import path

from .views import ActivityLogView, view_traceback_html

urlpatterns = [
    path("activity_log/", ActivityLogView.as_view()),
    path("activity_log/<int:log_id>/", view_traceback_html, name="activity_log"),
]

検証用のエンドポイント作成

ログが記録されることを検証するapiエンドポイントを作成します。
Middlewareで/log/から始まるパス以外を記録しているので、/api/から始まるAPIを検証用として作ります。
※検証用なので適当な実装ですが、クエリパラメータに?status=500を付けると意図的に500エラーを出すようにしています。

1. apiアプリの作成

$ python manage.py startapp api

2. 検証用のviewを作成

api/views.py
from rest_framework.decorators import api_view
from rest_framework.response import Response


@api_view(["GET"])
def view_sample(request):
    """
    sample view
    """
    response_status_code = request.query_params.get("status", "200")

    if response_status_code == "500":
        # 500 error(Name Error)
        aaa

    return Response({"message": "status 200"}, status=200)

3. urls.pyの作成

api/urls.py
from django.urls import path

from .views import view_sample

urlpatterns = [
    path("", view_sample),
]

記録されるログのイメージ

こんな感じに記録されます。id=6のログは500エラーなので、traceback_urlにURLが返ってきます。

traceback_urlを踏むとこんな感じのDjangoのエラー画面が出ます。

Discussion