profiling.py

# /usr/bin/python3
"""Profiling decorator"""
import inspect
import os
import threading
import time
from functools import wraps
from queue import Queue
from typing import Callable


PROFILE = True
PROFILING_T0: float = time.perf_counter()
PROFILING_EVENTS_QUEUE: Queue = Queue()  # [Tuple[str, str, float, float]]


def get_function_name(func: Callable):
    """Get file and function name"""
    module_name = func.__module__.split('.')[-1]
    if module_name == "__main__":
        module_name = os.path.basename(inspect.getfile(func))
    return f"{module_name}::{func.__name__}"


def push_profiling_event(name: str,
                         start_time: int,
                         end_time: int,
                         thread_id: Optional[str] = None):
    """Push profiling event"""
    if thread_id is None:
        thread_id = threading.current_thread().name
    # Queue is thread-safe
    PROFILING_EVENTS_QUEUE.put((name, thread_id,
                                start_time-PROFILING_T0,
                                end_time-PROFILING_T0))


def profile(func: Callable):
    """Profiling decorator"""
    @wraps(func)
    def wrapper(*args, **kwargs):
        if not PROFILE:
            return func(*args, **kwargs)

        start_time = time.perf_counter()
        retval = func(*args, **kwargs)
        end_time = time.perf_counter()

        push_profiling_event(get_function_name(func), start_time, end_time)
        return retval
    return wrapper