Loading docker-compose.test.yaml +4 −4 Original line number Diff line number Diff line version: '3.7' version: "3.7" services: package: Loading @@ -10,8 +10,8 @@ services: - ./test_envfile #command: ["sh", "-c", "pytest --cov=common /test/test.py"] #command: ["sh", "-c", "test-db-conn && pytest -v --cov=common /test/test.py"] command: ["sh", "-c", "test-db-conn && python3 /test/test.py"] #command: tail -f /dev/null #command: ["sh", "-c", "test-db-conn && python3 /test/test.py"] command: tail -f /dev/null depends_on: - postgres Loading src/common/logz.py +174 −20 Original line number Diff line number Diff line #!/usr/bin/env python3 # -*- coding: utf-8 -*- import os import sys """Provide a standard logger for all use cases.""" import logging # standard python imports import os from rich.logging import RichHandler from rich.traceback import install def create_logger(file_out=None, mode=None, encoding=None): def create_handler(file_out, mode, encoding, another_handler=None): """ create_logger: Creates a logger for all uses create_handler: Creates a logging handler to format text written by Python logging module. :param file_out: Path to file to output logs to, defaults to None :type file_out: str, optional Loading @@ -20,27 +18,56 @@ def create_logger(file_out=None, mode=None, encoding=None): :type mode: str, optional :param encoding: Encoding to open the file with, defaults to None :type encoding: str, optional :return: The logger that was created :rtype: Logger :param another_handler: Handler from another logging module, defaults to None :type another_handler: handler, optional :return: A Python loggging handler :rtype: handler """ # Remove any handler's that may have been set in the logging root for handler in logging.root.handlers[:]: handler.close() logging.root.removeHandler(handler) install() log_level = os.environ.get("LOGLEVEL", "INFO").upper() rich_handler = RichHandler(rich_tracebacks=True, markup=True) if file_out is not None: file_handler = logging.FileHandler(file_out, mode, encoding) file_handler_fmt = logging.Formatter( "[%(asctime)s]" + "%(levelname)8s - " + " - %(message)s" ) file_handler.setFormatter(file_handler_fmt) handlers = [rich_handler, file_handler] handlers = [file_handler] if another_handler: handlers = [another_handler, file_handler] else: file_handler = None handlers = [rich_handler] if another_handler: handlers = [another_handler] else: handlers = [] return file_handler, handlers def create_rich_logger(file_out=None, mode=None, encoding=None): """ create_rich_logger: Creates a Rich logger for all uses :param file_out: Path to file to output logs to, defaults to None :type file_out: str, optional :param mode: Mode to open the file with, defaults to None :type mode: str, optional :param encoding: Encoding to open the file with, defaults to None :type encoding: str, optional :return: A Rich logger instance :rtype: Logger """ try: from rich.logging import RichHandler from rich.traceback import install except ImportError: raise ImportError("Failed to load rich logger") install() log_level = os.environ.get("LOGLEVEL", "INFO").upper() rich_handler = RichHandler(rich_tracebacks=True, markup=True) file_handler, handlers = create_handler(file_out, mode, encoding, rich_handler) logging.basicConfig( level=log_level, format="%(message)s", Loading @@ -53,3 +80,130 @@ def create_logger(file_out=None, mode=None, encoding=None): file_handler.close() rich_handler.close() return logging.getLogger("rich") def create_loguru_logger(file_out=None, mode=None, encoding=None): """ create_loguru_logger: Creates a Rich logger for all uses :param file_out: Path to file to output logs to, defaults to None :type file_out: str, optional :param mode: Mode to open the file with, defaults to None :type mode: str, optional :param encoding: Encoding to open the file with, defaults to None :type encoding: str, optional :return: A Loguru logger instance :rtype: Logger """ try: from loguru import logger as loguru_logger except: print("Failed to load loguru logger") return loguru_logger.remove() log_level = os.environ.get("LOGLEVEL", "INFO").upper() if file_out is not None: loguru_logger.add( file_out, mode=mode, encoding=encoding, level=log_level, format="[{time:YYYY/MM/DD HH:mm:ss}] - {level} - {message}", backtrace=True, diagnose=True ) else: loguru_logger.add( sys.stderr, level=log_level, format="[{time:YYYY/MM/DD HH:mm:ss}] - {level} - {message}", backtrace=True, diagnose=True ) return loguru_logger def create_python_logger(file_out=None, mode=None, encoding=None): """ create_python_logger: Creates a Python logger for all uses :param file_out: Path to file to output logs to, defaults to None :type file_out: str, optional :param mode: Mode to open the file with, defaults to None :type mode: str, optional :param encoding: Encoding to open the file with, defaults to None :type encoding: str, optional :return: Python logger instance :rtype: Logger """ log_level = os.environ.get("LOGLEVEL", "INFO").upper() file_handler, handlers = create_handler(file_out, mode, encoding) stdout_handler = logging.StreamHandler(sys.stdout) stdout_handler_fmt = logging.Formatter( "[%(asctime)s]" + "%(levelname)8s - " + " - %(message)s" ) stdout_handler.setFormatter(stdout_handler_fmt) handlers.append(stdout_handler) if file_handler: logging.basicConfig( level=log_level, format="%(message)s", datefmt="[%Y/%m/%d %H:%M;%S]", handlers=handlers, ) else: logging.basicConfig( level=log_level, format="%(asctime)s" + "%(levelname)8s - " + " - %(message)s", datefmt="[%Y/%m/%d %H:%M;%S]" ) # if there is a file_handler set, close it before leaving :) # This prevents leaving open files if file_handler is not None: file_handler.close() return logging.getLogger() def create_logger(file_out=None, mode=None, encoding=None, logger_type="rich"): """ create_logger: Creates a logger for all uses :param file_out: Path to file to output logs to, defaults to None :type file_out: str, optional :param mode: Mode to open the file with, defaults to None :type mode: str, optional :param encoding: Encoding to open the file with, defaults to None :type encoding: str, optional :param logger_type: Logger to use for logging purpose, defaults to rich (rich (default), loguru, logging) :type logger_type: str, optional :return: The logger that was created :rtype: Logger """ # Remove any handler's that may have been set in the logging root for handler in logging.root.handlers[:]: handler.close() logging.root.removeHandler(handler) logger_type = os.getenv("LOGGER_TYPE", None) or logger_type mode = mode or "a" encoding = encoding or "utf-8" if logger_type == "loguru": return create_loguru_logger(file_out, mode, encoding) elif logger_type == "logging": return create_python_logger(file_out, mode, encoding) return create_rich_logger(file_out, mode, encoding) src/pyproject.toml +2 −1 Original line number Diff line number Diff line Loading @@ -30,7 +30,8 @@ classifiers = [ "Topic :: Software Development :: Build Tools" ] dependencies = [ "rich==13.7.1" "rich==13.7.1", "loguru==0.7.2" ] [project.optional-dependencies] Loading src/requirements.txt +2 −1 Original line number Diff line number Diff line Loading @@ -4,3 +4,4 @@ psycopg2-binary==2.9.9 pymssql==2.2.11 SQLAlchemy~=2.0.23 influxdb==5.3.1 loguru==0.7.2 No newline at end of file test/test_logz.py 0 → 100644 +36 −0 Original line number Diff line number Diff line #!/usr/bin/env python3 # -*- coding: utf-8 -*- import pytest import os import io import sys from common.logz import create_logger from common.file_operations import path_exists, read_file, delete_file @pytest.fixture(autouse=True) def cleanup_log_files(): log_files = ["test_rich.log", "test_loguru.log", "test_python_logging.log"] yield for file in log_files: if (path_exists(file)): delete_file(file) @pytest.mark.parametrize("logger_type, log_file", [ ("rich", "test_rich.log"), ("logger", "test_loguru.log"), ("logging", "test_python_logging.log") ]) def test_loggers(logger_type, log_file, capsys): logger = create_logger(file_out=log_file, logger_type=logger_type) custom_text = f"Test {logger_type} logger" logger.info(custom_text) captured = capsys.readouterr() stdout_output = captured.out assert custom_text in stdout_output + 'adadda', f"Expected '{custom_text}' in stdout, but got {stdout_output}" file_contents = read_file(log_file) if file_contents == None: return f"An error occured with file {log_file} for {logger_type} logger" assert custom_text in file_contents, f"Expected '{custom_text}' in log file, but got {file_contents}" Loading
docker-compose.test.yaml +4 −4 Original line number Diff line number Diff line version: '3.7' version: "3.7" services: package: Loading @@ -10,8 +10,8 @@ services: - ./test_envfile #command: ["sh", "-c", "pytest --cov=common /test/test.py"] #command: ["sh", "-c", "test-db-conn && pytest -v --cov=common /test/test.py"] command: ["sh", "-c", "test-db-conn && python3 /test/test.py"] #command: tail -f /dev/null #command: ["sh", "-c", "test-db-conn && python3 /test/test.py"] command: tail -f /dev/null depends_on: - postgres Loading
src/common/logz.py +174 −20 Original line number Diff line number Diff line #!/usr/bin/env python3 # -*- coding: utf-8 -*- import os import sys """Provide a standard logger for all use cases.""" import logging # standard python imports import os from rich.logging import RichHandler from rich.traceback import install def create_logger(file_out=None, mode=None, encoding=None): def create_handler(file_out, mode, encoding, another_handler=None): """ create_logger: Creates a logger for all uses create_handler: Creates a logging handler to format text written by Python logging module. :param file_out: Path to file to output logs to, defaults to None :type file_out: str, optional Loading @@ -20,27 +18,56 @@ def create_logger(file_out=None, mode=None, encoding=None): :type mode: str, optional :param encoding: Encoding to open the file with, defaults to None :type encoding: str, optional :return: The logger that was created :rtype: Logger :param another_handler: Handler from another logging module, defaults to None :type another_handler: handler, optional :return: A Python loggging handler :rtype: handler """ # Remove any handler's that may have been set in the logging root for handler in logging.root.handlers[:]: handler.close() logging.root.removeHandler(handler) install() log_level = os.environ.get("LOGLEVEL", "INFO").upper() rich_handler = RichHandler(rich_tracebacks=True, markup=True) if file_out is not None: file_handler = logging.FileHandler(file_out, mode, encoding) file_handler_fmt = logging.Formatter( "[%(asctime)s]" + "%(levelname)8s - " + " - %(message)s" ) file_handler.setFormatter(file_handler_fmt) handlers = [rich_handler, file_handler] handlers = [file_handler] if another_handler: handlers = [another_handler, file_handler] else: file_handler = None handlers = [rich_handler] if another_handler: handlers = [another_handler] else: handlers = [] return file_handler, handlers def create_rich_logger(file_out=None, mode=None, encoding=None): """ create_rich_logger: Creates a Rich logger for all uses :param file_out: Path to file to output logs to, defaults to None :type file_out: str, optional :param mode: Mode to open the file with, defaults to None :type mode: str, optional :param encoding: Encoding to open the file with, defaults to None :type encoding: str, optional :return: A Rich logger instance :rtype: Logger """ try: from rich.logging import RichHandler from rich.traceback import install except ImportError: raise ImportError("Failed to load rich logger") install() log_level = os.environ.get("LOGLEVEL", "INFO").upper() rich_handler = RichHandler(rich_tracebacks=True, markup=True) file_handler, handlers = create_handler(file_out, mode, encoding, rich_handler) logging.basicConfig( level=log_level, format="%(message)s", Loading @@ -53,3 +80,130 @@ def create_logger(file_out=None, mode=None, encoding=None): file_handler.close() rich_handler.close() return logging.getLogger("rich") def create_loguru_logger(file_out=None, mode=None, encoding=None): """ create_loguru_logger: Creates a Rich logger for all uses :param file_out: Path to file to output logs to, defaults to None :type file_out: str, optional :param mode: Mode to open the file with, defaults to None :type mode: str, optional :param encoding: Encoding to open the file with, defaults to None :type encoding: str, optional :return: A Loguru logger instance :rtype: Logger """ try: from loguru import logger as loguru_logger except: print("Failed to load loguru logger") return loguru_logger.remove() log_level = os.environ.get("LOGLEVEL", "INFO").upper() if file_out is not None: loguru_logger.add( file_out, mode=mode, encoding=encoding, level=log_level, format="[{time:YYYY/MM/DD HH:mm:ss}] - {level} - {message}", backtrace=True, diagnose=True ) else: loguru_logger.add( sys.stderr, level=log_level, format="[{time:YYYY/MM/DD HH:mm:ss}] - {level} - {message}", backtrace=True, diagnose=True ) return loguru_logger def create_python_logger(file_out=None, mode=None, encoding=None): """ create_python_logger: Creates a Python logger for all uses :param file_out: Path to file to output logs to, defaults to None :type file_out: str, optional :param mode: Mode to open the file with, defaults to None :type mode: str, optional :param encoding: Encoding to open the file with, defaults to None :type encoding: str, optional :return: Python logger instance :rtype: Logger """ log_level = os.environ.get("LOGLEVEL", "INFO").upper() file_handler, handlers = create_handler(file_out, mode, encoding) stdout_handler = logging.StreamHandler(sys.stdout) stdout_handler_fmt = logging.Formatter( "[%(asctime)s]" + "%(levelname)8s - " + " - %(message)s" ) stdout_handler.setFormatter(stdout_handler_fmt) handlers.append(stdout_handler) if file_handler: logging.basicConfig( level=log_level, format="%(message)s", datefmt="[%Y/%m/%d %H:%M;%S]", handlers=handlers, ) else: logging.basicConfig( level=log_level, format="%(asctime)s" + "%(levelname)8s - " + " - %(message)s", datefmt="[%Y/%m/%d %H:%M;%S]" ) # if there is a file_handler set, close it before leaving :) # This prevents leaving open files if file_handler is not None: file_handler.close() return logging.getLogger() def create_logger(file_out=None, mode=None, encoding=None, logger_type="rich"): """ create_logger: Creates a logger for all uses :param file_out: Path to file to output logs to, defaults to None :type file_out: str, optional :param mode: Mode to open the file with, defaults to None :type mode: str, optional :param encoding: Encoding to open the file with, defaults to None :type encoding: str, optional :param logger_type: Logger to use for logging purpose, defaults to rich (rich (default), loguru, logging) :type logger_type: str, optional :return: The logger that was created :rtype: Logger """ # Remove any handler's that may have been set in the logging root for handler in logging.root.handlers[:]: handler.close() logging.root.removeHandler(handler) logger_type = os.getenv("LOGGER_TYPE", None) or logger_type mode = mode or "a" encoding = encoding or "utf-8" if logger_type == "loguru": return create_loguru_logger(file_out, mode, encoding) elif logger_type == "logging": return create_python_logger(file_out, mode, encoding) return create_rich_logger(file_out, mode, encoding)
src/pyproject.toml +2 −1 Original line number Diff line number Diff line Loading @@ -30,7 +30,8 @@ classifiers = [ "Topic :: Software Development :: Build Tools" ] dependencies = [ "rich==13.7.1" "rich==13.7.1", "loguru==0.7.2" ] [project.optional-dependencies] Loading
src/requirements.txt +2 −1 Original line number Diff line number Diff line Loading @@ -4,3 +4,4 @@ psycopg2-binary==2.9.9 pymssql==2.2.11 SQLAlchemy~=2.0.23 influxdb==5.3.1 loguru==0.7.2 No newline at end of file
test/test_logz.py 0 → 100644 +36 −0 Original line number Diff line number Diff line #!/usr/bin/env python3 # -*- coding: utf-8 -*- import pytest import os import io import sys from common.logz import create_logger from common.file_operations import path_exists, read_file, delete_file @pytest.fixture(autouse=True) def cleanup_log_files(): log_files = ["test_rich.log", "test_loguru.log", "test_python_logging.log"] yield for file in log_files: if (path_exists(file)): delete_file(file) @pytest.mark.parametrize("logger_type, log_file", [ ("rich", "test_rich.log"), ("logger", "test_loguru.log"), ("logging", "test_python_logging.log") ]) def test_loggers(logger_type, log_file, capsys): logger = create_logger(file_out=log_file, logger_type=logger_type) custom_text = f"Test {logger_type} logger" logger.info(custom_text) captured = capsys.readouterr() stdout_output = captured.out assert custom_text in stdout_output + 'adadda', f"Expected '{custom_text}' in stdout, but got {stdout_output}" file_contents = read_file(log_file) if file_contents == None: return f"An error occured with file {log_file} for {logger_type} logger" assert custom_text in file_contents, f"Expected '{custom_text}' in log file, but got {file_contents}"