Stanislav Hejny
129 lines
4.5 KiB
Python
129 lines
4.5 KiB
Python
import inspect
|
|
import logging
|
|
import os
|
|
import sys
|
|
import threading
|
|
import traceback
|
|
from importlib.metadata import version
|
|
|
|
# Get version from installed package metadata
|
|
__version__ = version("amq_adapter") # Name must match pyproject.toml [project] name
|
|
__verbose__ = os.environ.get("AMQ_VERBOSE_LOGGING", "0") == "1"
|
|
|
|
|
|
def get_context_info():
|
|
"""
|
|
Retrieves the current module, line number, and function name.
|
|
|
|
Returns:
|
|
A tuple containing (module name, line number, function name).
|
|
Returns (None, None, None) if it fails to retrieve the information.
|
|
"""
|
|
try:
|
|
# Get the frame object for the current function call
|
|
frame = inspect.currentframe()
|
|
if frame is not None:
|
|
# Get the frame object for the caller of the current function
|
|
frame = frame.f_back.f_back
|
|
if frame:
|
|
code_context = frame.f_code
|
|
module_name = code_context.co_filename
|
|
module_name = (
|
|
"amqp" + module_name.split("/amqp", 1)[-1]
|
|
if "/amqp" in module_name
|
|
else module_name
|
|
)
|
|
line_number = frame.f_lineno
|
|
function_name = code_context.co_name
|
|
return module_name, line_number, function_name
|
|
else:
|
|
return None, None, None
|
|
except Exception:
|
|
return None, None, None
|
|
|
|
|
|
def logging_error(msg: str, *args) -> None:
|
|
"""
|
|
Log an error message according to current logging for 'ERROR' level.
|
|
Add module name, line and function name where the error occurred, on single line.
|
|
Args:
|
|
msg (str): The error message to log.
|
|
"""
|
|
_exec_info = sys.exc_info()[2]
|
|
_thread_id = threading.get_ident()
|
|
if _exec_info:
|
|
_tb = traceback.extract_tb(sys.exc_info()[2])[-1]
|
|
logging.error(
|
|
f"{_thread_id}:{__version__} [E] '{msg}' ---> '{_tb.filename}':{_tb.lineno}, '{_tb.name}()'",
|
|
*args,
|
|
)
|
|
else:
|
|
module, line_number, function_name = get_context_info()
|
|
if module and line_number and function_name:
|
|
logging.error(
|
|
f"{_thread_id}:{__version__} [E] '{msg}' ---> '{module}':{line_number}, '{function_name}()'",
|
|
*args,
|
|
)
|
|
else:
|
|
logging.error(f"{_thread_id}:{__version__} [E] {msg}", *args)
|
|
|
|
|
|
def logging_warning(msg: str, *args) -> None:
|
|
"""
|
|
Log a warning message according to current logging for 'WARN' level.
|
|
Add module name, line and function name where the print occurred, on single line.
|
|
Args:
|
|
msg (str): The warning message to log.
|
|
"""
|
|
_thread_id = threading.get_ident()
|
|
if not __verbose__:
|
|
logging.warning(f"{_thread_id}:{__version__} [W] {msg}", *args)
|
|
return
|
|
module, line_number, function_name = get_context_info()
|
|
if module and line_number and function_name:
|
|
logging.warning(
|
|
f"{_thread_id}:{__version__} [W] '{msg}' ---> '{module}':{line_number}, '{function_name}()'",
|
|
*args,
|
|
)
|
|
else:
|
|
logging.warning(f"{_thread_id}:{__version__} [W] {msg}", *args)
|
|
|
|
|
|
def logging_info(msg: str, *args) -> None:
|
|
"""
|
|
Log an info message according to current logging for 'INFO' level.
|
|
Add module name, line and function name where the print occurred, on single line.
|
|
Args:
|
|
msg (str): The info message to log.
|
|
"""
|
|
_thread_id = threading.get_ident()
|
|
if not __verbose__:
|
|
logging.info(f"{_thread_id}:{__version__} [*] {msg}", *args)
|
|
return
|
|
module, line_number, function_name = get_context_info()
|
|
if module and line_number and function_name:
|
|
logging.info(
|
|
f"{_thread_id}:{__version__} [*] '{msg}' ---> '{module}':{line_number}, '{function_name}()'",
|
|
*args,
|
|
)
|
|
else:
|
|
logging.info(f"{_thread_id}:{__version__} [*] {msg}", *args)
|
|
|
|
|
|
def logging_debug(msg: str, *args) -> None:
|
|
"""
|
|
Log a debug message according to current logging for 'DEBUG' level.
|
|
Add module name, line and function name where the print occurred, on single line.
|
|
Args:
|
|
msg (str): The debug message to log.
|
|
"""
|
|
_thread_id = threading.get_ident()
|
|
module, line_number, function_name = get_context_info()
|
|
if module and line_number and function_name:
|
|
logging.debug(
|
|
f"{_thread_id}:{__version__} [DBG] '{msg}' ---> '{module}':{line_number}, '{function_name}()'",
|
|
*args,
|
|
)
|
|
else:
|
|
logging.debug(f"{_thread_id}:{__version__} [DBG] {msg}", *args)
|