Python爬虫入门教程：从零开始抓取网页数据 - 程序员客栈

你是否需要一个关于Django中间件实现日志记录的完整案例

访客 全栈框架 1

本文目录导读：

  1. 项目结构
  2. 日志配置 (logger/logger_config.py)
  3. 中间件实现 (middleware/logging_middleware.py)
  4. settings.py配置
  5. 视图示例 (views.py)
  6. URL配置 (urls.py)
  7. 使用示例
  8. 日志输出示例

我来为你提供一个完整的Django中间件日志记录案例。

项目结构

log_middleware_demo/
├── log_middleware_demo/
│   ├── __init__.py
│   ├── settings.py
│   ├── urls.py
│   └── wsgi.py
├── middleware/
│   ├── __init__.py
│   └── logging_middleware.py
├── logger/
│   ├── __init__.py
│   └── logger_config.py
└── manage.py

日志配置 (logger/logger_config.py)

import os
import logging
import logging.handlers
from datetime import datetime
class LoggerConfig:
    def __init__(self, log_dir='logs'):
        # 创建日志目录
        self.log_dir = log_dir
        if not os.path.exists(log_dir):
            os.makedirs(log_dir)
    def setup_logging(self):
        """配置日志系统"""
        # 1. 访问日志配置
        access_logger = logging.getLogger('access_log')
        access_logger.setLevel(logging.INFO)
        # 访问日志处理器（按天轮转）
        access_handler = logging.handlers.TimedRotatingFileHandler(
            filename=os.path.join(self.log_dir, 'access.log'),
            when='midnight',
            interval=1,
            backupCount=30,
            encoding='utf-8'
        )
        access_format = logging.Formatter(
            '%(asctime)s - %(levelname)s - %(message)s'
        )
        access_handler.setFormatter(access_format)
        access_logger.addHandler(access_handler)
        # 2. 错误日志配置
        error_logger = logging.getLogger('error_log')
        error_logger.setLevel(logging.ERROR)
        error_handler = logging.handlers.RotatingFileHandler(
            filename=os.path.join(self.log_dir, 'error.log'),
            maxBytes=10*1024*1024,  # 10MB
            backupCount=5,
            encoding='utf-8'
        )
        error_format = logging.Formatter(
            '%(asctime)s - %(levelname)s - %(name)s - %(message)s'
        )
        error_handler.setFormatter(error_format)
        error_logger.addHandler(error_handler)
        # 3. SQL查询日志配置
        sql_logger = logging.getLogger('sql_log')
        sql_logger.setLevel(logging.DEBUG)
        sql_handler = logging.handlers.TimedRotatingFileHandler(
            filename=os.path.join(self.log_dir, 'sql.log'),
            when='H',  # 每小时轮转
            interval=1,
            backupCount=24,
            encoding='utf-8'
        )
        sql_format = logging.Formatter(
            '%(asctime)s - %(duration)ss - %(sql)s'
        )
        sql_handler.setFormatter(sql_format)
        sql_logger.addHandler(sql_handler)
        # 4. 性能日志配置
        performance_logger = logging.getLogger('performance_log')
        performance_logger.setLevel(logging.WARNING)
        perf_handler = logging.handlers.TimedRotatingFileHandler(
            filename=os.path.join(self.log_dir, 'performance.log'),
            when='midnight',
            interval=1,
            backupCount=7,
            encoding='utf-8'
        )
        perf_format = logging.Formatter(
            '%(asctime)s - %(path)s - %(method)s - %(duration).4fs - %(status_code)s'
        )
        perf_handler.setFormatter(perf_format)
        performance_logger.addHandler(perf_handler)
        # 控制台输出（可选）
        console_handler = logging.StreamHandler()
        console_format = logging.Formatter(
            '%(asctime)s - %(levelname)s - %(message)s'
        )
        console_handler.setFormatter(console_format)
        console_handler.setLevel(logging.INFO)
        # 为所有logger添加控制台处理器
        for logger in [access_logger, error_logger, sql_logger, performance_logger]:
            logger.addHandler(console_handler)
        return {
            'access': access_logger,
            'error': error_logger,
            'sql': sql_logger,
            'performance': performance_logger
        }

中间件实现 (middleware/logging_middleware.py)

import time
import json
import traceback
from datetime import datetime
from django.utils.deprecation import MiddlewareMixin
from django.http import HttpRequest, HttpResponse
from django.conf import settings
class RequestLoggingMiddleware(MiddlewareMixin):
    """请求日志中间件"""
    def __init__(self, get_response=None):
        super().__init__(get_response)
        self.loggers = None
    def process_request(self, request):
        """请求处理前记录"""
        # 记录请求开始时间
        request.start_time = time.time()
        # 记录请求信息
        request_info = {
            'method': request.method,
            'path': request.path,
            'query_params': dict(request.GET),
            'user': str(request.user) if request.user.is_authenticated else 'anonymous',
            'ip': self.get_client_ip(request),
            'user_agent': request.META.get('HTTP_USER_AGENT', ''),
            'referer': request.META.get('HTTP_REFERER', ''),
        }
        # 记录请求体（POST/PUT/PATCH）
        if request.method in ['POST', 'PUT', 'PATCH']:
            try:
                if request.content_type == 'application/json':
                    request_info['body'] = json.loads(request.body)
                else:
                    request_info['body'] = request.POST.dict()
            except (ValueError, AttributeError):
                request_info['body'] = 'Unable to parse body'
        # 存储请求信息以便后续使用
        request.request_info = request_info
        # 日志记录
        self.log_request(request, request_info)
    def process_response(self, request, response):
        """响应处理后记录"""
        # 计算请求处理时间
        if hasattr(request, 'start_time'):
            duration = time.time() - request.start_time
            # 记录访问日志
            self.log_access(request, response, duration)
            # 记录性能日志（对慢请求进行告警）
            if duration > 1.0:  # 超过1秒的请求
                self.log_performance_warning(request, response, duration)
            # 记录错误日志（对错误响应进行记录）
            if response.status_code >= 400:
                self.log_error(request, response, duration)
        return response
    def process_exception(self, request, exception):
        """异常处理"""
        self.log_exception(request, exception)
    def log_request(self, request, request_info):
        """记录请求日志"""
        logger = self.get_logger('access')
        if logger:
            logger.info(f"Request: {json.dumps(request_info, indent=2, ensure_ascii=False)}")
    def log_access(self, request, response, duration):
        """记录访问日志"""
        logger = self.get_logger('access')
        if logger:
            log_data = {
                'timestamp': datetime.now().isoformat(),
                'method': request.method,
                'path': request.path,
                'status_code': response.status_code,
                'duration': f'{duration:.4f}s',
                'ip': self.get_client_ip(request),
                'user': str(request.user) if request.user.is_authenticated else 'anonymous',
                'response_size': len(response.content) if hasattr(response, 'content') else 0
            }
            logger.info(f"Access: {json.dumps(log_data, ensure_ascii=False)}")
    def log_error(self, request, response, duration):
        """记录错误日志"""
        logger = self.get_logger('error')
        if logger:
            error_data = {
                'path': request.path,
                'method': request.method,
                'status_code': response.status_code,
                'duration': f'{duration:.4f}s',
                'ip': self.get_client_ip(request),
                'user': str(request.user) if request.user.is_authenticated else 'anonymous',
                'response_reason': response.reason_phrase if hasattr(response, 'reason_phrase') else ''
            }
            logger.error(f"Error Response: {json.dumps(error_data, ensure_ascii=False)}")
    def log_exception(self, request, exception):
        """记录异常日志"""
        logger = self.get_logger('error')
        if logger:
            error_data = {
                'path': request.path,
                'method': request.method,
                'error_type': type(exception).__name__,
                'error_message': str(exception),
                'ip': self.get_client_ip(request),
                'user': str(request.user) if request.user.is_authenticated else 'anonymous',
                'traceback': traceback.format_exc()
            }
            logger.error(f"Exception: {json.dumps(error_data, indent=2, ensure_ascii=False)}")
    def log_performance_warning(self, request, response, duration):
        """记录性能告警"""
        logger = self.get_logger('performance')
        if logger:
            perf_data = {
                'path': request.path,
                'method': request.method,
                'duration': duration,
                'status_code': response.status_code,
                'timestamp': datetime.now().isoformat()
            }
            logger.warning(f"Slow Request: {json.dumps(perf_data, ensure_ascii=False)}")
    def get_logger(self, log_type):
        """获取日志记录器"""
        if self.loggers is None:
            try:
                from logger.logger_config import LoggerConfig
                config = LoggerConfig()
                self.loggers = config.setup_logging()
            except Exception as e:
                # 如果日志配置失败，使用默认日志
                import logging
                self.loggers = {
                    'access': logging.getLogger('django'),
                    'error': logging.getLogger('django'),
                    'sql': logging.getLogger('django'),
                    'performance': logging.getLogger('django')
                }
        return self.loggers.get(log_type)
    def get_client_ip(self, request):
        """获取客户端IP地址"""
        x_forwarded_for = request.META.get('HTTP_X_FORWARDED_FOR')
        if x_forwarded_for:
            ip = x_forwarded_for.split(',')[0]
        else:
            ip = request.META.get('REMOTE_ADDR')
        return ip
class SQLQueryLoggingMiddleware(MiddlewareMixin):
    """SQL查询日志中间件"""
    def process_request(self, request):
        """启动SQL查询记录"""
        from django.db import connection
        request.initial_queries = len(connection.queries)
    def process_response(self, request, response):
        """记录SQL查询信息"""
        from django.db import connection
        if hasattr(request, 'initial_queries'):
            queries = connection.queries[request.initial_queries:]
            if queries and settings.DEBUG:
                logger = self.get_sql_logger()
                if logger:
                    for query in queries:
                        log_data = {
                            'time': datetime.now().isoformat(),
                            'sql': query['sql'],
                            'duration': query['time']
                        }
                        logger.debug(f"SQL Query: {json.dumps(log_data, ensure_ascii=False)}")
        return response
    def get_sql_logger(self):
        """获取SQL日志记录器"""
        try:
            from logger.logger_config import LoggerConfig
            config = LoggerConfig()
            loggers = config.setup_logging()
            return loggers.get('sql')
        except:
            return None
class UserActivityLoggingMiddleware(MiddlewareMixin):
    """用户活动日志中间件"""
    def process_request(self, request):
        """记录用户活动开始"""
        request.activity_start_time = time.time()
    def process_view(self, request, view_func, view_args, view_kwargs):
        """在视图处理前记录"""
        # 记录视图信息
        request.view_info = {
            'view_name': view_func.__name__ if hasattr(view_func, '__name__') else str(view_func),
            'view_module': view_func.__module__ if hasattr(view_func, '__module__') else '',
            'view_args': view_args,
            'view_kwargs': view_kwargs
        }
    def process_response(self, request, response):
        """记录用户活动"""
        if hasattr(request, 'activity_start_time') and hasattr(request, 'user'):
            duration = time.time() - request.activity_start_time
            user = request.user
            if user.is_authenticated:
                activity_data = {
                    'user': user.username,
                    'email': user.email,
                    'path': request.path,
                    'method': request.method,
                    'duration': duration,
                    'view_name': getattr(request, 'view_info', {}).get('view_name', 'unknown')
                }
                # 这里可以记录到数据库或日志
                self.log_user_activity(activity_data)
        return response
    def log_user_activity(self, activity_data):
        """记录用户活动到日志"""
        import logging
        logger = logging.getLogger('django.user_activity')
        logger.info(f"User Activity: {json.dumps(activity_data, ensure_ascii=False)}")

settings.py配置

# settings.py
# 添加中间件
MIDDLEWARE = [
    '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',
    # 自定义日志中间件
    'middleware.logging_middleware.RequestLoggingMiddleware',
    'middleware.logging_middleware.SQLQueryLoggingMiddleware',
    'middleware.logging_middleware.UserActivityLoggingMiddleware',
]
# 日志配置（Django默认日志配置增强）
LOGGING = {
    'version': 1,
    'disable_existing_loggers': False,
    'formatters': {
        'verbose': {
            'format': '{levelname} {asctime} {module} {message}',
            'style': '{',
        },
        'simple': {
            'format': '{levelname} {message}',
            'style': '{',
        },
    },
    'handlers': {
        'console': {
            'level': 'INFO',
            'class': 'logging.StreamHandler',
            'formatter': 'simple',
        },
        'file': {
            'level': 'INFO',
            'class': 'logging.handlers.TimedRotatingFileHandler',
            'filename': 'logs/django.log',
            'when': 'midnight',
            'backupCount': 30,
            'formatter': 'verbose',
        },
    },
    'loggers': {
        'django': {
            'handlers': ['console', 'file'],
            'level': 'INFO',
            'propagate': True,
        },
        'django.request': {
            'handlers': ['file'],
            'level': 'ERROR',
            'propagate': False,
        },
        'django.user_activity': {
            'handlers': ['file'],
            'level': 'INFO',
            'propagate': True,
        },
    },
}
# 调试模式下记录SQL查询
if DEBUG:
    LOGGING['loggers']['django.db.backends'] = {
        'handlers': ['console'],
        'level': 'DEBUG',
        'propagate': False,
    }

视图示例 (views.py)

from django.http import JsonResponse
from django.views.decorators.csrf import csrf_exempt
import time
def index(request):
    """正常请求"""
    return JsonResponse({"message": "Hello World"})
def slow_request(request):
    """模拟慢请求"""
    time.sleep(2)  # 模拟2秒的延迟
    return JsonResponse({"message": "Slow request completed"})
def error_request(request):
    """模拟错误请求"""
    # 模拟一个错误
    raise ValueError("This is a test error")
@csrf_exempt
def post_example(request):
    """POST请求示例"""
    if request.method == 'POST':
        data = {
            'method': request.method,
            'path': request.path,
            'data': request.POST.dict()
        }
        return JsonResponse(data)
    return JsonResponse({"error": "Method not allowed"}, status=405)

URL配置 (urls.py)

from django.urls import path
from . import views
urlpatterns = [
    path('', views.index, name='index'),
    path('slow/', views.slow_request, name='slow'),
    path('error/', views.error_request, name='error'),
    path('post/', views.post_example, name='post'),
]

使用示例

# 测试代码
import requests
# 测试正常请求
response = requests.get('http://localhost:8000/')
print(f"Status: {response.status_code}")
# 测试慢请求
response = requests.get('http://localhost:8000/slow/')
print(f"Slow request status: {response.status_code}")
# 测试错误请求
try:
    response = requests.get('http://localhost:8000/error/')
except:
    print("Error occurred as expected")
# 测试POST请求
response = requests.post('http://localhost:8000/post/', data={'key': 'value'})
print(f"POST response: {response.json()}")

日志输出示例

access.log

2024-01-15 10:30:45,123 - INFO - Request: {
  "method": "GET",
  "path": "/",
  "query_params": {},
  "user": "anonymous",
  "ip": "127.0.0.1",
  "user_agent": "python-requests/2.28.1",
  "referer": ""
}
2024-01-15 10:30:45,456 - INFO - Access: {
  "timestamp": "2024-01-15T10:30:45.456",
  "method": "GET",
  "path": "/",
  "status_code": 200,
  "duration": "0.0234s",
  "ip": "127.0.0.1",
  "user": "anonymous",
  "response_size": 27
}

performance.log

2024-01-15 10:31:00,789 - WARNING - Slow Request: {
  "path": "/slow/",
  "method": "GET",
  "duration": 2.0123,
  "status_code": 200,
  "timestamp": "2024-01-15T10:31:00.789"
}

error.log

2024-01-15 10:31:15,234 - ERROR - Exception: {
  "path": "/error/",
  "method": "GET",
  "error_type": "ValueError",
  "error_message": "This is a test error",
  "ip": "127.0.0.1",
  "user": "anonymous",
  "traceback": "Traceback (most recent call last):\n  ..."
}

这个完整的日志中间件案例提供了：

  1. 请求/响应日志：记录所有HTTP请求的详细信息
  2. 性能监控：检测慢请求并记录告警
  3. 错误日志：记录异常和错误响应
  4. 用户活动追踪：记录已认证用户的操作
  5. SQL查询记录：记录数据库查询（调试模式下）
  6. 灵活的配置：支持按天轮转、大小限制、多个日志文件

你可以根据需要调整日志级别、格式和存储策略。

标签： Django中间件 日志记录

文章来源： 访客
版权声明：除非特别标注，否则均为本站原创文章，转载时请以链接形式注明文章出处。

上一篇怎样通过一个文件上传案例对比Flask和FastAPI的处理方式

下一篇如何用Python全栈框架实现WebSocket双向通信

相关文章

抱歉，评论功能暂时关闭!