Logging decorators

_logging_decorators.py

import logging
import os
from subprocess import CompletedProcess

import subprocess


def uvicorn_logger():
    return logging.getLogger('uvicorn')


def log(level: int = logging.INFO):
    def decorator(func):
        def wrapper(*args, **kwargs):
            uvicorn_logger().log(level, f'{func.__name__} called with args: {args}, kwargs: {kwargs}')
            return func(*args, **kwargs)
        return wrapper
    return decorator


def log_execution(result: CompletedProcess) -> CompletedProcess:
    uvicorn_logger().log(logging.INFO, f'`{result.args}` returned with returncode: {result.returncode}')
    if result.returncode != 0:
        uvicorn_logger().warning('there were some errors')
        uvicorn_logger().log(logging.INFO, f'stdout:\n{result.stdout.decode("utf-8")}')
        uvicorn_logger().log(logging.INFO, f'stderr:\n{result.stderr.decode("utf-8")}')
        raise subprocess.CalledProcessError(result.returncode, result.args)
    return result


def log_and_execute(command: str, *args, **kwargs):
    preset_arguments = ('shell', 'check', 'capture_output')
    if any(preset in kwargs for preset in preset_arguments):
        uvicorn_logger().error(f'preset arguments {preset_arguments} will be overwritten')
        raise ValueError(f'preset arguments {preset_arguments} will be overwritten')
    kwargs.update({'shell': True, 'check': False, 'capture_output': True})
    cwd = kwargs.get('cwd', os.getcwd())
    uvicorn_logger().log(logging.INFO, f'`{command}` called with args: {args}, kwargs: {kwargs} in {cwd}')
    return log_execution(subprocess.run(command, *args, **kwargs))

example.py

    @log()
    def generate_patches(self, project: Project) -> list[Patch]:
        config_path = self._generate_config_for_project(project)
        log_and_execute('mvn clean', cwd=project.path.source)
        tool_execution = log_and_execute(f'java -jar {self.tool_executable_jar_path} {config_path}')
        if tool_execution.returncode != 0:
            log_and_execute('mvn -f ./dep-pom.xml dependency:copy-dependencies', cwd=project.path.source)
            log_and_execute(f'java -jar {self.tool_executable_jar_path} {config_path}')

        patch_path = os.path.join(project.path.measurement, 'results', 'patches')
        if os.path.isdir(patch_path):
            return [
                Patch(file_path=os.path.join(patch_path, file))
                for file in os.listdir(patch_path) if file.endswith('.diff')
            ]
        return []