from django.db import connections from enum import Enum from typing import List, Optional, Set, Tuple, Union import logging import functools import inspect import os import time logger = logging.getLogger(__name__) class InfoAppel(Enum): """ infos supplémentaires possibles """ # ajouté au nom de logger CLASSE = 'classe' # ajouté dans le message PID = 'pid' def get_nom_module(func) -> str: mod = inspect.getmodule(func) return mod.__name__ if mod else None def get_nombre_requetes() -> int: """ renvoie le nombre total de requêtes (il peut y avoir plusieurs connexions) """ return sum(len(c.queries) for c in connections.all()) def get_nom_classe(inclure: Union[ List[InfoAppel], Set[InfoAppel], Tuple[InfoAppel] ] = (), *args) -> str: """ renvoie le nom de classe correspondant à l'appel de fonction """ nom_classe = None if InfoAppel.CLASSE in inclure and args: classe = getattr(args[0], '__class__', None) nom_classe = getattr(classe, '__name__') if classe else None return nom_classe def get_logger(nom_module: Optional[str], nom_classe: Optional[str]) -> logging.Logger: """ renvoie le logger correspondant au module et à la classe """ return logging.getLogger(f'{nom_module}.{nom_classe}' if nom_module and nom_classe else nom_classe or nom_module) def duree_execution(avertir_apres: int = None, inclure: Union[List[InfoAppel], Set[InfoAppel], Tuple[InfoAppel]] = ()): """ décorateur pour tracer le temps d'exécution d'une fonction :param avertir_apres: durée (en ms) au-delà de laquelle on souhaite un log d'avertissement, defaults to None :type avertir_apres: int, optional :param inclure: infos supplémentaires, defaults to () :type inclure: Union[List[InfoAppel], Set[InfoAppel], Tuple[InfoAppel]], optional :return: résultat de la fonction d'origine """ def inner(func): nom_module = get_nom_module(func) @functools.wraps(func) def wrapper(*args, **kwargs): temps_debut = time.time() try: resultat = func(*args, **kwargs) finally: try: temps_ecoule = round((time.time() - temps_debut) * 1000) func_logger = get_logger(nom_module, get_nom_classe(inclure, *args)) log_level = logging.WARNING if isinstance(avertir_apres, int) and temps_ecoule > avertir_apres else logging.DEBUG func_logger.log(log_level, "%s'%s' exécuté en %d ms", '[%s] ' % os.getpid() if InfoAppel.PID in inclure else '', func.__name__, temps_ecoule) except Exception: logger.exception('impossible de tracer la durée') return resultat return wrapper return inner def nombre_requetes(avertir_apres: int = None, inclure: Union[List[InfoAppel], Set[InfoAppel], Tuple[InfoAppel]] = ()): """ décorateur pour tracer le nombre de requêtes d'une fonction :param avertir_apres: nombre de requêtes au-delà duquel on souhaite un log d'avertissement, defaults to None :type avertir_apres: int, optional :param inclure: infos supplémentaires, defaults to () :type inclure: Union[List[InfoAppel], Set[InfoAppel], Tuple[InfoAppel]], optional :return: résultat de la fonction d'origine """ def inner(func): nom_module = get_nom_module(func) @functools.wraps(func) def wrapper(*args, **kwargs): nb_debut = get_nombre_requetes() try: resultat = func(*args, **kwargs) finally: try: nb_requetes = get_nombre_requetes() - nb_debut func_logger = get_logger(nom_module, get_nom_classe(inclure, *args)) log_level = logging.WARNING if isinstance(avertir_apres, int) and nb_requetes > avertir_apres else logging.DEBUG func_logger.log(log_level, "%s'%s' a exécuté %d requête(s)", '[%s] ' % os.getpid() if InfoAppel.PID in inclure else '', func.__name__, nb_requetes) except Exception: logger.exception('impossible de tracer le nombre de requêtes') return resultat return wrapper return inner