refactor: Change root file pit.lib

README.md

@@ -26,7 +26,7 @@ pip install log-call
 Then, import the `log_call` decorator in your Python script:
 
 ```python
-from pit.lib.log_call.log_decorator import log_call
+from pit.log_call.log_decorator import log_call
 ```
 
 ## Usage
@@ -35,7 +35,7 @@ Decorate your function using `@log_call`:
 
 ```python
 import logging
-from pit.lib.log_call.log_decorator import log_call
+from pit.log_call.log_decorator import log_call
 
 
 @log_call(level=logging.INFO, log_args=True, log_response=True)
@@ -55,8 +55,7 @@ For asynchronous functions, simply apply the decorator as you would with a synch
 ```python
 import logging
 
-frompit.lib.log_call.log_decorator
-import log_call
+from pit.log_call.log_decorator import log_call
 
 
 @log_call(level=logging.INFO, log_args=True, log_response=True)
@@ -93,7 +92,7 @@ In case of an exception, the error will be logged and then re-raised.
 ````python
 
 import logging
-from pit.lib.log_call.log_decorator import log_call
+from pit.log_call.log_decorator import log_call
 
 logger = logging.getLogger(__name__)
 

src/pit/log_call/log_decorator.py

+import asyncio
+import functools
+import logging
+import random
+import string
+
+start_message = "Start span:[%s] method:[%s]"
+args_message = " with args [%s]"
+end_message = "End   span:[%s] method:[%s]"
+result_message = " result [%s]"
+exception_message = " exception [%s]"
+
+__alphabet = string.ascii_letters + string.digits
+
+
+def get_spam(length: int = 16) -> str:
+    """
+    Generate a random string with numbers and letters
+    :param length: The length of the string
+    :return: A random string with numbers and letters
+    """
+    return "".join(random.choices(__alphabet, k=length))
+
+
+def log_error(entry_point, logger, level, spam_id, e):
+    if logger.isEnabledFor(level):
+        logger.log(
+            level, end_message + exception_message, spam_id, entry_point, repr(e)
+        )
+
+
+def log_end(entry_point, log_response, is_verbose, logger, level, spam_id, result):
+    if logger.isEnabledFor(level):
+        should_log_response = log_response if log_response is not None else is_verbose
+        if should_log_response:
+            logger.log(
+                level, end_message + result_message, spam_id, entry_point, repr(result)
+            )
+        else:
+            logger.log(level, end_message, spam_id, entry_point)
+
+
+def log_start(entry_point, log_args, is_verbose, logger, level, spam_id, args, kwargs):
+    if logger.isEnabledFor(level):
+        should_log_args = log_args if log_args is not None else is_verbose
+        if should_log_args:
+            args_repr = [repr(a) for a in args]
+            kwargs_repr = [f"{k}={v!r}" for k, v in kwargs.items()]
+            signature = ", ".join(args_repr + kwargs_repr)
+            logger.log(
+                level, start_message + args_message, spam_id, entry_point, signature
+            )
+        else:
+            logger.log(level, start_message, spam_id, entry_point)
+
+
+def log_call(
+    _func=None,
+    _logger: logging.Logger | str | None = None,
+    level: int = logging.DEBUG,
+    log_args: bool | None = None,
+    log_response: bool | None = None,
+):
+    """
+    Decorator to log the start and end of a function call, including its arguments and result.
+    :param _logger: logger where the logs will be written, if None the logger of the module will be used
+    :param _func: the function to decorate
+    :param level: logging level to use, default is logging.DEBUG
+    :param log_args: whether to log the function arguments, default is True if the logger is in DEBUG mode
+        , for higher log levels this value have to be set explicitly
+    :param log_response: whether to log the function result, default is True if the logger is in DEBUG mode
+    , for higher log levels this value have to be set explicitly
+    :return: the decorated function
+    """
+
+    def log_decorator_info(__func):
+
+        def decorate_sync_async(___func):
+            if _logger is not None:
+                if isinstance(_logger, str):
+                    logger = logging.getLogger(_logger)
+                else:
+                    logger = _logger
+            else:
+                logger = logging.getLogger(___func.__module__)
+            spam_id = get_spam()
+            entry_point = ___func.__qualname__
+            is_verbose = level == logging.DEBUG
+            if asyncio.iscoroutinefunction(___func):
+
+                async def decorated(*args, **kwargs):
+                    log_start(
+                        entry_point,
+                        log_args,
+                        is_verbose,
+                        logger,
+                        level,
+                        spam_id,
+                        args,
+                        kwargs,
+                    )
+
+                    try:
+                        result = await ___func(*args, **kwargs)
+                        log_end(
+                            entry_point,
+                            log_response,
+                            is_verbose,
+                            logger,
+                            level,
+                            spam_id,
+                            result,
+                        )
+
+                        return result
+                    except Exception as e:
+                        log_error(entry_point, logger, level, spam_id, e)
+                        raise e
+
+            else:
+
+                def decorated(*args, **kwargs):
+                    log_start(
+                        entry_point,
+                        log_args,
+                        is_verbose,
+                        logger,
+                        level,
+                        spam_id,
+                        args,
+                        kwargs,
+                    )
+
+                    try:
+                        result = ___func(*args, **kwargs)
+                        log_end(
+                            entry_point,
+                            log_response,
+                            is_verbose,
+                            logger,
+                            level,
+                            spam_id,
+                            result,
+                        )
+
+                        return result
+                    except Exception as e:
+                        log_error(entry_point, logger, level, spam_id, e)
+                        raise e
+
+            return functools.wraps(___func)(decorated)
+
+        return decorate_sync_async(__func)
+
+    if _func is None:
+        # El decorador se utilizó con paréntesis, por lo que devolvemos el decorador real
+        return log_decorator_info
+    else:
+        # El decorador se utilizó sin paréntesis, por lo que decoramos la función directamente
+        return log_decorator_info(_func)

tests/pit/log_call/test_log_decorator.py

+import logging
+
+import pytest
+
+from pit.log_call.log_decorator import (
+    start_message,
+    log_call,
+    end_message,
+    args_message,
+    result_message,
+    exception_message,
+)
+
+
+@log_call(level=logging.INFO)
+def div(a: float, b: float) -> float:
+    return a / b
+
+
+@log_call(level=logging.INFO, log_args=True, log_response=True)
+def div2(a: float, b: float) -> float:
+    return a / b
+
+
+@log_call(level=logging.INFO, log_args=True, log_response=True)
+def do_nothing():
+    pass
+
+
+@log_call(level=logging.DEBUG)
+def debug_function():
+    pass
+
+
+@log_call(level=logging.DEBUG, log_args=False, log_response=False)
+def debug_no_args():
+    pass
+
+
+@log_call(level=logging.INFO, log_args=True, log_response=True)
+async def async_function():
+    pass
+
+
+@log_call(level=logging.INFO, log_args=False, log_response=False)
+async def async_function_noargs():
+    pass
+
+
+@log_call
+def no_arguments() -> None:
+    pass
+
+
+class Calculator:
+    @log_call(level=logging.INFO)
+    def div(self, a: float, b: float) -> float:
+        return a / b
+
+
+def test_function_not_log_args(caplog):
+    caplog.set_level(logging.INFO)
+    assert div(1, 2) == 0.5
+    with pytest.raises(ZeroDivisionError):
+        div(1, 0)
+
+    assert caplog.records[0].msg == start_message
+    assert caplog.records[0].args[1] == "div"
+    assert caplog.records[1].msg == end_message
+    assert caplog.records[2].msg == start_message
+    assert caplog.records[3].msg == end_message + exception_message
+
+
+def test_function_log_args(caplog):
+    caplog.set_level(logging.INFO)
+    assert div2(a=10, b=2) == 5
+
+    assert caplog.records[0].msg == start_message + args_message
+    assert caplog.records[0].args[1] == "div2"
+    assert "a=10" in caplog.records[0].args[2]
+    assert "b=2" in caplog.records[0].args[2]
+    assert caplog.records[1].msg == end_message + result_message
+    assert "5" in caplog.records[1].args[2]
+
+
+def test_function_void_with_no_args_log_args_is_none(caplog):
+    caplog.set_level(logging.INFO)
+    do_nothing()
+    assert caplog.records[0].msg == start_message + args_message
+    assert caplog.records[0].args[1] == "do_nothing"
+    assert caplog.records[1].msg == end_message + result_message
+    assert "None" in caplog.records[1].args[2]
+
+
+def test_debug_function_log_args_as_default(caplog):
+    caplog.set_level(logging.DEBUG)
+    debug_function()
+    assert caplog.records[0].msg == start_message + args_message
+    assert caplog.records[0].args[1] == "debug_function"
+    assert caplog.records[1].msg == end_message + result_message
+    assert "None" in caplog.records[1].args[2]
+
+
+def test_debug_function_not_log_args(caplog):
+    caplog.set_level(logging.DEBUG)
+    debug_no_args()
+    assert caplog.records[0].msg == start_message
+    assert caplog.records[0].args[1] == "debug_no_args"
+    assert caplog.records[1].msg == end_message
+
+
+def test_not_log_with_low_log_level(caplog):
+    caplog.set_level(logging.INFO)
+    debug_no_args()
+    assert len(caplog.records) == 0
+
+
+def test_log_class_method(caplog):
+    caplog.set_level(logging.DEBUG)
+    calculator = Calculator()
+    assert calculator.div(10, 2) == 5
+
+    assert caplog.records[0].msg == start_message
+    assert caplog.records[0].args[1] == "Calculator.div"
+    assert caplog.records[1].msg == end_message
+
+
+@pytest.mark.asyncio
+async def test_log_async_function(caplog):
+    caplog.set_level(logging.INFO)
+    await async_function()
+    assert caplog.records[0].msg == start_message + args_message
+    assert caplog.records[0].args[1] == "async_function"
+    assert caplog.records[1].msg == end_message + result_message
+    assert "None" in caplog.records[1].args[2]
+
+
+@pytest.mark.asyncio
+async def test_log_async_function_no_args(caplog):
+    caplog.set_level(logging.INFO)
+    await async_function_noargs()
+    assert caplog.records[0].msg == start_message
+    assert caplog.records[0].args[1] == "async_function_noargs"
+    assert caplog.records[1].msg == end_message
+
+
+def test_log_no_arguments(caplog):
+    caplog.set_level(logging.DEBUG)
+    no_arguments()
+    assert caplog.records[0].msg == start_message + args_message
+    assert caplog.records[0].args[1] == "no_arguments"
+    assert caplog.records[1].msg == end_message + result_message
+    assert "None" in caplog.records[1].args[2]
+
+
+logger = logging.getLogger(__name__)
+logger_decorator = logging.getLogger("decorator_logger")
+
+
+@log_call(
+    level=logging.DEBUG, log_args=False, log_response=False, _logger="decorator_logger"
+)
+def sum_function(a, b):
+    logger.log(logging.INFO, f"Summing {a} and {b}")
+    return a + b
+
+
+@log_call(
+    level=logging.DEBUG, log_args=False, log_response=False, _logger=logger_decorator
+)
+def sum_function2(a, b):
+    logger.log(logging.INFO, f"Summing {a} and {b}")
+    return a + b
+
+
+def test_with_str_logger(caplog):
+    caplog.set_level(logging.DEBUG)
+    res = sum_function(5, 6)
+    assert res == 11
+    assert caplog.records[0].name == "decorator_logger"
+    assert caplog.records[1].name == "tests.pit.log_call.test_log_decorator"
+    assert caplog.records[2].name == "decorator_logger"
+
+
+def test_with_logger_instance_logger(caplog):
+    caplog.set_level(logging.DEBUG)
+    res = sum_function2(5, 6)
+    assert res == 11
+    assert caplog.records[0].name == "decorator_logger"
+    assert caplog.records[0].levelname == "DEBUG"
+    assert caplog.records[1].name == "tests.pit.log_call.test_log_decorator"
+    assert caplog.records[1].levelname == "INFO"
+    assert caplog.records[2].name == "decorator_logger"