#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ # @Version : Python 3.11.4 # @Software : Sublime Text 4 # @Author : StudentCWZ # @Email : StudentCWZ@outlook.com # @Date : 2023/10/28 17:44 # @File : logger.py # @Description : """ import os import sys from flask import Flask, request, g from loguru import logger from .format import patching from application.libs.helper import ConfigHelper class FlaskLoguru: def __init__(self, app=None): """ Construct a new FlaskLoguru instance. :param app: Optional Flask app to initialize this FlaskLoguru instance for. """ if app is not None: self.init_app(app) def init_app(self, app: Flask) -> None: """ Initialize the given Flask app for logging. :param app: The Flask app to initialize. """ # Initialize ConfigHelper config_helper = ConfigHelper(app) # Fetch the logging configuration from the app's config cfg = config_helper.Logger # Configure logger based on provided config log_type = cfg.get('Type', 'terminal') level = cfg.get('Level', 'DEBUG') path = cfg.get('Path') name = cfg.get('Name') _format = cfg.get('Format') or patching rotation = cfg.get('Rotation', '1 day') enqueue = cfg.get('Enqueue', False) retention = cfg.get('Retention', '10 days') # Remove any existing handlers from the logger logger.remove() # If the logger type is 'both' or 'file' and a path and a name are provided, add a file logger if self._should_add_file_logger(log_type, path, name): full_path = os.path.join(path, name) logger.add(full_path, level=level, format=_format, rotation=rotation, retention=retention, enqueue=enqueue) # If the logger type is 'both' or 'terminal', add a terminal logger if self._should_add_terminal_logger(log_type): logger.add(sys.stderr, level=level, format=_format) # Register this logger with the app app.extensions.setdefault('loguru', {}) app.extensions['loguru'][self] = logger # Add logger to Flask's before, after and teardown request handlers @app.before_request def _log_request_start(): """ Log the start of the request. """ if request.method == 'POST': data = dict(url=request.url, method=request.method, ip=request.remote_addr, request_body=request.get_json()) else: data = dict(url=request.url, method=request.method, ip=request.remote_addr, request_body="") g.logger = logger.bind(data=data) g.logger.info('Request started') @app.after_request def _log_request_completion(response): """ Log the end of the request. """ g.logger.info('Request completed') return response @app.teardown_request def _log_request_teardown(exception=None): """ Log any exception that occurred during the request. """ if exception: data = dict(exception=str(exception)) g.logger = logger.bind(data=data) g.logger.exception('An error occurred during the request') @staticmethod def _should_add_file_logger(log_type: str, path: str, name: str) -> bool: """ Determine if a file logger should be added. """ return log_type in ['both', 'file'] and path and name @staticmethod def _should_add_terminal_logger(log_type: str) -> bool: """ Determine if a terminal logger should be added. """ return log_type in ['both', 'terminal']