Django REST framework 日志（重写drf_api_logger）

超级小的大杯柠檬水

发布于 2023-07-09 10:37:53

3830

发布于 2023-07-09 10:37:53

文章被收录于专栏：CYCY

Django REST framework 日志

默认的drf-api-logger没有保存用户并且没有获取日志的接口本文通过重写drf-api-logger增加访问用户及获取日志的接口并且增加定时器删除日志

drf_api_logger

文档

优点：您可以将 API 信息记录到数据库中或侦听不同用例的记录器信号，也可以同时执行这两项操作。记录器使用单独的线程来运行，因此不会影响 API 响应时间。

安装

pip install drf-api-logger

配置

# settings.py
INSTALLED_APPS = [
	'...',
    'drf_api_logger',
]

MIDDLEWARE = [
	'...',
    'drf_api_logger.middleware.api_logger_middleware.APILoggerMiddleware',
]

# drf-api-logger 配置
DRF_API_LOGGER_DATABASE = True # 是否记录到数据库
DRF_API_LOGGER_SIGNAL = True # 是否发送信号
DRF_API_LOGGER_PATH_TYPE = 'ABSOLUTE' # 路径类型
DRF_API_LOGGER_SKIP_URL_NAME = [] # 跳过的url
DRF_API_LOGGER_SKIP_NAMESPACE = [] # 跳过的命名空间(应用程序)
DRF_API_LOGGER_METHODS = [] # 跳过的方法
DRF_API_LOGGER_STATUS_CODES = [] # 跳过的状态码
DRF_API_LOGGER_EXCLUDE_KEYS = ['password', 'token', 'access', 'refresh','AUTHORIZATION'] # 在日志中隐藏敏感数据
DRF_LOGGER_QUEUE_MAX_SIZE = 50 # 日志队列最大长度
DRF_LOGGER_QUEUE_FLUSH_INTERVAL = 10 # 日志队列刷新间隔

查看

可以在 Django 管理面板的管理面板查看。

重写

drf_api_logger其实就是django的一个应用复制drf_api_logger的源码后执行 pip uninstall drf-api-logger 删除模块(不删也没事)

添加用户信息

在模型中添加用户字段

# models.py
from django.contrib.auth import get_user_model
User = get_user_model()
···
class APILogsModel(BaseModel):
	···
	user = models.ForeignKey(User,null=True, blank=True, on_delete=models.CASCADE,verbose_name="用户",help_text="用户")
	···

在中间件中修改添加数据库时的方法()

# middleware\api_logger_middleware.py
# 导入你的验证Token方法，我使用的是Django-Rest-Knox
from knox.auth import TokenAuthentication

class APILoggerMiddleware:
···
	def __call__(self, request):
 		···
		# 获取用户模型
		try:
		    user,_ = TokenAuthentication().authenticate(request)
		except Exception as e: 
		    user = None
		
		data = dict(
		  user=user,
		  api=mask_sensitive_data(api, mask_api_parameters=True),
		  headers=mask_sensitive_data(headers),
		  body=mask_sensitive_data(request_data),
		  method=method,
		  client_ip_address=get_client_ip(request),
		  response=mask_sensitive_data(response_body),
		  status_code=response.status_code,
		  execution_time=time.time() - start_time,
		  added_on=timezone.now()
		)
		···

添加获取日志的接口

编写序列化器

# serializers.py
from .models import APILogsModel
from rest_framework import serializers
import json

class Jsonserializer(serializers.CharField):
    """编写一个序列化字段,将字符串转为json对象"""
    def to_representation(self, value):
        try:
            return json.loads(value)
        except Exception as e:
            return value
        

class APILogsSerializers(serializers.ModelSerializer):
    # 将json字符串转为json对象，方便前端展示，否则前端展示的是字符串

    headers = Jsonserializer()
    body = Jsonserializer()
    response = Jsonserializer()
    
    class Meta:
        model = APILogsModel
        fields = '__all__'

编写视图

# views.py
from .serializers import APILogsSerializers
from .models import APILogsModel
from rest_framework import viewsets
from rest_framework import permissions

class APILogsViewSet(viewsets.ReadOnlyModelViewSet):
    """添加获取日志列表的视图函数"""
    queryset = APILogsModel.objects.all()
    serializer_class = APILogsSerializers
    permission_classes = [permissions.IsAdminUser]

编写路由

# urls.py
from django.urls import path
from rest_framework import routers
router = routers.DefaultRouter()
from . import views
router.register("log",views.APILogsViewSet,"log")
urlpatterns = []
urlpatterns += router.urls

完整代码

# models.py
# middleware\api_logger_middleware.py
from django.db import models

from drf_api_logger.utils import database_log_enabled
from django.contrib.auth import get_user_model

User = get_user_model()

if database_log_enabled():
    """
    Load models only if DRF_API_LOGGER_DATABASE is True
    """
    class BaseModel(models.Model):
        id = models.BigAutoField(primary_key=True)

        added_on = models.DateTimeField()

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

        class Meta:
            abstract = True
            ordering = ('-added_on',)


    class APILogsModel(BaseModel):
        api = models.CharField(max_length=1024, help_text='API URL')
        user = models.ForeignKey(User,null=True, blank=True, on_delete=models.CASCADE,verbose_name="用户",help_text="用户")
        headers = models.TextField()
        body = models.TextField()
        method = models.CharField(max_length=10, db_index=True)
        client_ip_address = models.CharField(max_length=50)
        response = models.TextField()
        status_code = models.PositiveSmallIntegerField(help_text='Response status code', db_index=True)
        execution_time = models.DecimalField(decimal_places=5, max_digits=8,
                                             help_text='Server execution time (Not complete response time.)')

        def __str__(self):
            return self.api

        class Meta:
            db_table = 'drf_api_logs'
            verbose_name = 'API Log'
            verbose_name_plural = 'API Logs'

import json
import time
import re
from django.conf import settings
from django.urls import resolve
from django.utils import timezone

from drf_api_logger import API_LOGGER_SIGNAL
from drf_api_logger.start_logger_when_server_starts import LOGGER_THREAD
from drf_api_logger.utils import get_headers, get_client_ip, mask_sensitive_data

from knox.auth import TokenAuthentication
from rest_framework import exceptions

"""
File: api_logger_middleware.py
Class: APILoggerMiddleware
重写以在日志中记录用户信息
"""


class APILoggerMiddleware:
    def __init__(self, get_response):
        self.get_response = get_response
        # One-time configuration and initialization.

        self.DRF_API_LOGGER_DATABASE = False
        if hasattr(settings, 'DRF_API_LOGGER_DATABASE'):
            self.DRF_API_LOGGER_DATABASE = settings.DRF_API_LOGGER_DATABASE

        self.DRF_API_LOGGER_SIGNAL = False
        if hasattr(settings, 'DRF_API_LOGGER_SIGNAL'):
            self.DRF_API_LOGGER_SIGNAL = settings.DRF_API_LOGGER_SIGNAL

        self.DRF_API_LOGGER_PATH_TYPE = 'ABSOLUTE'
        if hasattr(settings, 'DRF_API_LOGGER_PATH_TYPE'):
            if settings.DRF_API_LOGGER_PATH_TYPE in ['ABSOLUTE', 'RAW_URI', 'FULL_PATH']:
                self.DRF_API_LOGGER_PATH_TYPE = settings.DRF_API_LOGGER_PATH_TYPE

        self.DRF_API_LOGGER_SKIP_URL_NAME = []
        if hasattr(settings, 'DRF_API_LOGGER_SKIP_URL_NAME'):
            if type(settings.DRF_API_LOGGER_SKIP_URL_NAME) is tuple or type(
                    settings.DRF_API_LOGGER_SKIP_URL_NAME) is list:
                self.DRF_API_LOGGER_SKIP_URL_NAME = settings.DRF_API_LOGGER_SKIP_URL_NAME

        self.DRF_API_LOGGER_SKIP_NAMESPACE = []
        if hasattr(settings, 'DRF_API_LOGGER_SKIP_NAMESPACE'):
            if type(settings.DRF_API_LOGGER_SKIP_NAMESPACE) is tuple or type(
                    settings.DRF_API_LOGGER_SKIP_NAMESPACE) is list:
                self.DRF_API_LOGGER_SKIP_NAMESPACE = settings.DRF_API_LOGGER_SKIP_NAMESPACE

        self.DRF_API_LOGGER_METHODS = []
        if hasattr(settings, 'DRF_API_LOGGER_METHODS'):
            if type(settings.DRF_API_LOGGER_METHODS) is tuple or type(
                    settings.DRF_API_LOGGER_METHODS) is list:
                self.DRF_API_LOGGER_METHODS = settings.DRF_API_LOGGER_METHODS

        self.DRF_API_LOGGER_STATUS_CODES = []
        if hasattr(settings, 'DRF_API_LOGGER_STATUS_CODES'):
            if type(settings.DRF_API_LOGGER_STATUS_CODES) is tuple or type(
                    settings.DRF_API_LOGGER_STATUS_CODES) is list:
                self.DRF_API_LOGGER_STATUS_CODES = settings.DRF_API_LOGGER_STATUS_CODES

    def __call__(self, request):

        # Run only if logger is enabled.
        if self.DRF_API_LOGGER_DATABASE or self.DRF_API_LOGGER_SIGNAL:

            url_name = resolve(request.path_info).url_name
            namespace = resolve(request.path_info).namespace

            # Always skip Admin panel
            if namespace == 'admin':
                return self.get_response(request)

            # Skip for url name
            if url_name in self.DRF_API_LOGGER_SKIP_URL_NAME:
                return self.get_response(request)

            # Skip entire app using namespace
            if namespace in self.DRF_API_LOGGER_SKIP_NAMESPACE:
                return self.get_response(request)

            start_time = time.time()
            request_data = ''
            try:
                request_data = json.loads(request.body) if request.body else ''
            except:
                pass

            # Code to be executed for each request before
            # the view (and later middleware) are called.
            response = self.get_response(request)

            # Only log required status codes if matching
            if self.DRF_API_LOGGER_STATUS_CODES and response.status_code not in self.DRF_API_LOGGER_STATUS_CODES:
                return response

            # Code to be executed for each request/response after
            # the view is called.

            headers = get_headers(request=request)
            method = request.method

            # Log only registered methods if available.
            if len(self.DRF_API_LOGGER_METHODS) > 0 and method not in self.DRF_API_LOGGER_METHODS:
                return response

            if response.get('content-type') in ('application/json', 'application/vnd.api+json', 'application/gzip'):
                
                if response.get('content-type') == 'application/gzip':
                    response_body = '** GZIP Archive **'
                elif getattr(response, 'streaming', False):
                    response_body = '** Streaming **'
                else:
                    if type(response.content) == bytes:
                        response_body = json.loads(response.content.decode())
                    else:
                        response_body = json.loads(response.content)
                if self.DRF_API_LOGGER_PATH_TYPE == 'ABSOLUTE':
                    api = request.build_absolute_uri()
                elif self.DRF_API_LOGGER_PATH_TYPE == 'FULL_PATH':
                    api = request.get_full_path()
                elif self.DRF_API_LOGGER_PATH_TYPE == 'RAW_URI':
                    api = request.get_raw_uri()
                else:
                    api = request.build_absolute_uri()
                
                
                try:
                    user,_ = TokenAuthentication().authenticate(request)
                except Exception as e: 
                    user = None
                
                data = dict(
                    user=user,
                    api=mask_sensitive_data(api, mask_api_parameters=True),
                    headers=mask_sensitive_data(headers),
                    body=mask_sensitive_data(request_data),
                    method=method,
                    client_ip_address=get_client_ip(request),
                    response=mask_sensitive_data(response_body),
                    status_code=response.status_code,
                    execution_time=time.time() - start_time,
                    added_on=timezone.now()
                )
                if self.DRF_API_LOGGER_DATABASE:
                    if LOGGER_THREAD:
                        d = data.copy()
                        d['headers'] = json.dumps(d['headers'], indent=4, ensure_ascii=False)
                        if request_data:
                            d['body'] = json.dumps(d['body'], indent=4, ensure_ascii=False)
                        d['response'] = json.dumps(d['response'], indent=4, ensure_ascii=False)
                        LOGGER_THREAD.put_log_data(data=d)
                if self.DRF_API_LOGGER_SIGNAL:
                    API_LOGGER_SIGNAL.listen(**data)
            else:
                return response
        else:
            response = self.get_response(request)
        return response

本文参与腾讯云自媒体同步曝光计划，分享自作者个人站点/博客。

原始发表：2023-06-29，如有侵权请联系 cloudcommunity@tencent.com 删除

django