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 删除模块(不删也没事)
添加用户信息
在模型中添加用户字段文章来源：https://www.toymoban.com/news/detail-510180.html
# 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="用户")
···
在中间件中修改添加数据库时的方法()文章来源地址https://www.toymoban.com/news/detail-510180.html
# 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
到了这里，关于Django REST framework 日志（重写drf_api_logger）的文章就介绍完了。如果您还想了解更多内容，请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章，希望大家以后多多支持TOY模板网！
