Merge branch 'file_logging' into 'develop' (957778b8) · Commits · Wohlgemuth, Jason / common-package

docker-compose.test.yaml

+2 −0

Original line number	Diff line number	Diff line
		@@ -9,6 +9,8 @@ services:
		environment:
		- DATABASE_HOST=postgres
		- DATABASE_TIMEOUT=60
		- DATABASE_LOG_FILE=test.log
		- DATABASE_LOG_MODE=w
		- LOGLEVEL=debug
		command: ["sh", "-c", "python3 /test/test.py"]
		depends_on:

src/common/database.py

+34 −10

Original line number	Diff line number	Diff line
		@@ -30,6 +30,16 @@ class Database(ABC):
		- DATABASE_SCHEMA: a string representing the schema to default to
		when creating a database connection

		Log file info stored in system environment variables:
		- DATABASE_LOG_PATH: a string representing the file path to record
		all logged data. defaults to, ""
		- DATABASE_LOG_MODE: a string representing the mode to open the log
		file. Standard convention for file modes are
		used. defaults to, "a"
		- DATABASE_LOG_ENCODING: a string representing the type of encoding
		to use when handling the log file.
		defaults to, "utf-8"

		:param connection_info: a dictionary containing connection information
		derived from the environmental variables described above.

		@@ -37,6 +47,8 @@ class Database(ABC):
		than explicitly calling the `open` function, as database connection
		will be created and destroyed behind the scenes, preventing lingering
		database connections.

		:note: File logging is disabled by default.
		"""
		DEFAULT_TIMEOUT = ce('DATABASE_TIMEOUT', 60)
		DEFAULT_DB = ce('DATABASE_DB', 'postgres')
		@@ -46,12 +58,18 @@ class Database(ABC):
		DEFAULT_PORT = int(ce('DATABASE_PORT', 5432))
		DEFAULT_SCHEMA = ce('DATABASE_SCHEMA', 'public')
		DEFAULT_ENGINE = ce('DATABASE_ENGINE', 'postgresql')
		DEFAULT_LOG_PATH = ce('DATABASE_LOG_FILE', '')
		DEFAULT_LOG_MODE = ce('DATABASE_LOG_MODE', 'a')
		DEFAULT_LOG_ENCODING = ce('DATABASE_LOG_ENCODING', 'utf-8')
		# define a URI string if URI is perferred to connect
		DEFAULT_URI = (DEFAULT_ENGINE + '://' +
		DEFAULT_USER + ':' + str(DEFAULT_PW) + '@' +
		DEFAULT_HOST + '/' + DEFAULT_DB)

		def __init__(self, connection_info={'dbName': DEFAULT_DB,
		def __init__(self, log_file_info={'path': DEFAULT_LOG_PATH,
		'mode': DEFAULT_LOG_MODE,
		'encoding': DEFAULT_LOG_ENCODING},
		connection_info={'dbName': DEFAULT_DB,
		'dbUser': DEFAULT_USER,
		'dbPassword': DEFAULT_PW,
		'dbSchema': DEFAULT_SCHEMA,
		@@ -65,7 +83,12 @@ class Database(ABC):

		:param connection_info: a dictionary containing connection info
		"""
		if log_file_info['path'] == '':
		self.logger = create_logger()
		else:
		self.logger = create_logger(log_file_info['path'],
		log_file_info['mode'],
		log_file_info['encoding'])
		self.connection_info = connection_info
		self.logger.debug(connection_info)
		self.connection = None
		@@ -122,6 +145,7 @@ class Database(ABC):
		self.connection.close()
		self.connection = None


		def is_open(self):
		"""Determine if the database is open or not."""
		if self.cursor is not None and self.connection is not None:

src/common/logz.py

+34 −3

Original line number	Diff line number	Diff line
		@@ -8,12 +8,43 @@ from rich.logging import RichHandler
		from rich.traceback import install


		def create_logger():
		"""Create a logger for use in all cases."""
		def create_logger(file_out=None, mode=None, encoding=None):
		"""
		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
		: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)

		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]
		else:
		handlers = [rich_handler]
		logging.basicConfig(level=log_level, format='%(message)s',
		datefmt="[%Y/%m/%d %H:%M;%S]",
		handlers=[rich_handler])
		handlers=handlers)
		# 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()
		rich_handler.close()
		return logging.getLogger('rich')

test/test.py

+82 −5

Original line number	Diff line number	Diff line
		@@ -2,23 +2,100 @@
		# -- coding: utf-8 --
		"""Unit test common functionality."""
		import unittest
		import os
		from common.database import Database
		from common.mixins.postgres import PostgresMixin
		from common.logz import create_logger


		class DBPG(Database, PostgresMixin):
		"""Create the Database and PostgresMixin class."""
		pass


		class PostgresTestCase(unittest.TestCase):
		"""Test the Postgres connection case."""
		def setUp(self) -> None:
		return super().setUp()

		def test_open(self):
		with DBPG() as db:
		print(db.query("SELECT * FROM test_table"))
		"""
		test_open:
		Test that the DB can successfully be opened
		"""
		with DBPG() as data_base:
		print(data_base.query("SELECT * FROM test_table"))


		class LogzTestCase(unittest.TestCase):
		"""
		LogzTestCase:
		Test case for testing the Logz.py file. This class tests both,
		create_logger's use with a database and standalone.

		:param unittest: Define this class as an invididual unit of testing
		:type unittest: TestCase
		"""

		def test_create_logger(self):
		"""
		test_create_logger:
		Method to test the create_logger function
		"""
		# create a DB connection
		with DBPG() as data_base:
		# use the logger in the db to log some info
		data_base.logger.info("THIS IS TEST 1!")
		data_base.logger.info("THIS IS TEST 2!")
		data_base.logger.debug("THIS IS TEST 3!")
		data_base.logger.error("THIS IS TEST 4!")
		# verify that the log file is indeed a file
		self.assertTrue(os.path.isfile("./test.log"))
		# verify that the file exists
		self.assertTrue(os.path.exists("./test.log"))
		# get the size of the file
		file_size = os.path.getsize("./test.log")
		# verify that there is data within the file
		self.assertTrue(file_size > 0)

		# open the log file
		with open("./test.log", 'r', encoding="utf-8") as file:
		# create bools to track if the info was found in the file
		found_t1 = False
		found_t2 = False
		found_t3 = False
		found_t4 = False
		# iterate through the file's lines
		for line in file.readlines():
		# if any of the test logs are found, record their existance
		if "THIS IS TEST 1!" in line:
		found_t1 = True
		elif "THIS IS TEST 2!" in line and "INFO" in line:
		found_t2 = True
		elif "THIS IS TEST 3!" in line and "DEBUG" in line:
		found_t3 = True
		elif "THIS IS TEST 4!" in line and "ERROR" in line:
		found_t4 = True
		# validate that all the test logs were found
		self.assertTrue(found_t1 and found_t2 and found_t3 and found_t4)

		# create a path for the standalone log file
		path = "./standalone.log"
		# create a logger to test its abilities standalone
		logger = create_logger(path, 'a', "utf-8")
		# log some info
		logger.info("THIS IS TEST 5!")
		# verify that a the path is a file
		self.assertTrue(os.path.isfile(path))
		# verify that the file exists
		self.assertTrue(os.path.exists(path))
		# verify that there is data in the file
		file_size = os.path.getsize(path)
		self.assertGreater(file_size, 0)

		# verify that the data from the logger is in the file
		with open(path, 'r', encoding="utf-8") as file:
		line = file.readline()
		self.assertTrue("THIS IS TEST 5" in line)
		self.assertTrue("INFO" in line)
		file.close()


		if __name__ == "__main__":