Merge branch 'feature/db_logger' into 'develop' (736fc86f) · Commits · Wohlgemuth, Jason / common-package

src/common/mixins/README.md

0 → 100644

+17 −0

Original line number	Diff line number	Diff line
		The intent of the class PostgresLoggerMixin is to provide a quick and easy way to log information to both the terminal and the database. The class is of type Database from the common_package, and so leverages both the database and the logger that is created by it’s parent class, Database(), from common_package.

		All one needs to initiate the PostgresLogger class is a schema name, then they would call the log just like they would normally:

		from common.database import Database

		db_log = PostgresLogger(self.workspace)
		db_log.info("This is a log message", w=True)

		This will log an INFO level log message and attempt to write it to the database.

		Features of the this custom logging class:
		Passing the -w flag will log the message to the log table of a schema in the database.
		Passing the -c flag will add the log to an internal list with the intent being to dump all of the logs in the database later. For example, if you have a running loop and want to log absolutely everything, but don’t want to open and close a database connection through every iteration, simply pass the -c flag to your log method, and after your loop, run the write_log_collection_to_database(). This logs all messages in the list and clears the log collection, self.log_collection. Note that this can be leveraged without using the logger, just pass write_log_collection_to_database() a list of tuples representing log messages.

		NOTE:
		Note that logging provides overhead, so logging a lot of information in long running loops will slow down your functions. Just keep this in mind when deciding what you want to log.
		No newline at end of file

src/common/mixins/postgres.py

+1 −1

Original line number	Diff line number	Diff line
		@@ -43,7 +43,7 @@ class PostgresMixin():
		self.logger.error(f'Unknown type: {search_path}.')
		self.search_path = 'public'
		self.cursor.execute(f"""SET search_path TO {self.search_path};""")
		self.cursor.commit()
		self.connection.commit()
		self.logger.debug('Successfully set search_path')
		except psycopg2.OperationalError as error:
		self.logger.error(f'Database Error: {error}')

src/common/mixins/postgres_logger.py

0 → 100644

+118 −0

Original line number	Diff line number	Diff line
		from os import times
		import time

		from common.database import Database
		from common.mixins.postgres import PostgresMixin

		class PostgresLoggerMixin(Database, PostgresMixin):
		"""
		Log Handler for easy logging.

		The intent of this handler is to have a common place to easily print logs
		to the terminal and/or record them to the .log table of a schema. After
		initializing the class, passing the w parameter to the log function will tell
		the function to write the log to the database, for example:
		db_logger.info("TEST MESSAGE", w=True)

		Attributes:
		schema (str): The name of the schema

		"""
		def __init__(self, schema, **kwargs):
		"""
		The constructor for the PostgresLoggerMixin class.

		Parameters:
		schema (str): The name of the schema
		"""
		super().__init__()
		self.schema = schema
		self.log_collection = []
		self.logger = Database().logger

		def debug(self, message, w=False, c=False):
		"""Logs a debug message."""
		if w:
		self.check_table_exists()
		self.log_to_database('DEBUG', message)
		elif c:
		self.collect_logs('DEBUG', message)
		self.logger.debug(message)

		def info(self, message, w=False, c=False):
		if w:
		self.check_table_exists()
		self.log_to_database('INFO', message)
		elif c:
		self.collect_logs('INFO', message)
		self.logger.info(message)

		def warning(self, message, w=False, c=False):
		if w:
		self.check_table_exists()
		self.log_to_database('WARNING', message)
		elif c:
		self.collect_logs('WARNING', message)
		self.logger.warning(message)

		def error(self, message, w=False, c=False):
		"""Logs an error message."""
		if w:
		self.check_table_exists()
		self.log_to_database('ERROR', message)
		elif c:
		self.collect_logs('ERROR', message)
		self.logger.error(message)

		def critical(self, message, w=False, c=False):
		"""Logs a critical Message"""
		if w:
		self.check_table_exists()
		self.log_to_database('CRITICAL', message)
		elif c:
		self.collect_logs('CRITICAL', message)
		self.logger.critical(message)

		# Only call from an exception handler
		def exception(self, message, w=False, c=False):
		"""Logs and exception message."""
		if w:
		self.check_table_exists()
		self.log_to_database('EXCEPTION', message)
		elif c:
		self.collect_logs('EXCEPTION', message)
		self.logger.exception(message)

		def check_table_exists(self):
		"""Checks if log table exists for workspace."""
		self.open()
		self.cursor.execute(f"CREATE TABLE IF NOT EXISTS {self.schema}.log (levelname TEXT, message TEXT, ts timestamp default now());")
		self.close()

		def log_to_database(self, loglevel, message):
		"Logs log message to database."
		self.open()
		self.cursor.execute(f"INSERT INTO {self.schema}.log (levelname, message) VALUES('{loglevel}', '{message}');")
		self.close()

		def collect_logs(self, level, message):
		"""Appends log messages to list"""
		log = (level, message)
		self.log_collection.append(tuple(log))

		def write_log_collection_to_database(self, log_list=None):
		"""Writes collection of logs to database and empties log list"""
		self.check_table_exists()
		# This keeps us from mucking up the internal list, even though we double
		# the code
		self.open()
		if log_list:
		for l in log_list:
		self.cursor.execute(f"INSERT INTO {self.schema}.log VALUES('{l[0]}', '{l[1]}');")
		elif self.log_collection:
		for l in self.log_collection:
		self.cursor.execute(f"INSERT INTO {self.schema}.log VALUES('{l[0]}', '{l[1]}');")
		self.log_collection = []
		else:
		self.warning("No logs recorded.")
		self.close()

test/test.py

+100 −0

Original line number	Diff line number	Diff line
		@@ -5,8 +5,12 @@ import unittest
		import os
		import subprocess
		import logging
		import time
		from datetime import datetime
		from webbrowser import get
		from common.database import Database
		from common.mixins.postgres import PostgresMixin
		from common.mixins.postgres_logger import PostgresLoggerMixin
		from common.logz import create_logger


		@@ -120,6 +124,102 @@ class LogzTestCase(unittest.TestCase):
		log = create_logger()
		self.assertTrue(isinstance(log, logging.Logger))

		def test_postgresloggermixin(self):
		"""
		test_postgresloggermixin:
		Method to test the postgresloggermixin class.
		"""
		with DBPG() as data_base:
		data_base.cursor.execute("DROP TABLE IF EXISTS public.log")
		plm = PostgresLoggerMixin('public')
		self.assertTrue(isinstance(plm, PostgresLoggerMixin))

		# # Test log message
		# plm.info("Test Message")

		# # test writing log message to database
		# test_log = "Test log to db"

		plm.debug("Test debug log to db", w=True)
		plm.info("Test info log to db", w=True)
		plm.warning("Test warning log to db", w=True)
		plm.error("Test error log to db", w=True)
		plm.critical("Test critical log to db", w=True)
		plm.exception("Test exception log to db", w=True)

		with DBPG() as data_base:
		get_single_log = data_base.query("SELECT * FROM public.log")

		self.assertListEqual(
		[
		("DEBUG", "Test debug log to db", get_single_log[0][2]),
		("INFO", "Test info log to db", get_single_log[1][2]),
		("WARNING", "Test warning log to db", get_single_log[2][2]),
		("ERROR", "Test error log to db", get_single_log[3][2]),
		("CRITICAL", "Test critical log to db", get_single_log[4][2]),
		("EXCEPTION", "Test exception log to db", get_single_log[5][2])
		],
		[
		get_single_log[0],
		get_single_log[1],
		get_single_log[2],
		get_single_log[3],
		get_single_log[4],
		get_single_log[5]
		]
		)

		# Test writing a collection to database
		plm.debug("test debug collection", c=True)
		plm.info("test info collection", c=True)
		plm.warning("test warning collection", c=True)
		plm.error("test error collection", c=True)
		plm.critical("test critical collection", c=True)
		plm.exception("test exception collection", c=True)

		plm.write_log_collection_to_database()

		with DBPG() as data_base:
		get_logs = data_base.query("SELECT * FROM public.log")

		self.assertListEqual(
		[
		("DEBUG", "test debug collection", get_logs[6][2]),
		("INFO", "test info collection", get_logs[7][2]),
		("WARNING", "test warning collection", get_logs[8][2]),
		("ERROR", "test error collection", get_logs[9][2]),
		("CRITICAL", "test critical collection", get_logs[10][2]),
		("EXCEPTION", "test exception collection", get_logs[11][2])
		],
		[
		get_logs[6], get_logs[7], get_logs[8], get_logs[9], get_logs[10], get_logs[11]
		]
		)

		test_custom_list = [
		("INFO", "custom_list_message"),
		("WARNING", "Warning Custom"),
		("CRITICAL", "YO THIS IS A CRIT")
		]

		plm.write_log_collection_to_database(log_list=test_custom_list)

		with DBPG() as data_base:
		get_cust = data_base.query("SELECT * FROM public.log")

		self.assertListEqual(
		[
		("INFO", "custom_list_message", get_cust[12][2]),
		("WARNING", "Warning Custom", get_cust[13][2]),
		("CRITICAL", "YO THIS IS A CRIT", get_cust[14][2]),
		],
		[
		get_cust[12], get_cust[13], get_cust[14]
		]
		)

		# test empty list:
		plm.write_log_collection_to_database()

		if __name__ == "__main__":
		unittest.main(verbosity=2)