pipenv run format (8f1fd3b2) · Commits · GSHS Utilities / common-package

src/common/auth.py

+7 −4

Original line number	Diff line number	Diff line
		"""Provide authentication methods."""
		from typing import Dict, Any, Callable
		from urllib.parse import urlparse, urlencode

		import hashlib
		import os
		from typing import Any, Callable, Dict
		from urllib.parse import urlencode, urlparse

		from common.env import check_environment as ce

		try:
		from ldap3 import Server, Connection, ALL, SUBTREE
		from ldap3.core.exceptions import LDAPException, LDAPBindError
		from ldap3 import ALL, SUBTREE, Connection, Server
		from ldap3.core.exceptions import LDAPBindError, LDAPException
		except ImportError:
		import sys

		from common.logz import create_logger

		log = create_logger()
		log.warn("To use this module, install common-package[auth] extra.")
		sys.exit(1)

src/common/crud_table.py

+8 −24

Original line number	Diff line number	Diff line
		@@ -76,8 +76,7 @@ class CRUDTable(ABC):
		# assure that all columns are defined
		# FIXME - asserts should only be used inside tests
		assert all(arg in self.columns.keys() for arg in kwargs.keys()), (
		"Must supply values for all columns to create an entry. "
		+ f"Columns: {self.columns}"
		"Must supply values for all columns to create an entry. " + f"Columns: {self.columns}"
		)
		# get a 'pretty' string of column names
		column_names = str(list(kwargs.keys()))[1:-1].replace("'", "")
		@@ -88,9 +87,7 @@ class CRUDTable(ABC):
		f"VALUES ({', '.join(['%s']*len(kwargs.keys()))})"
		)
		# tell the user that we are executing an insert query
		self.logger.info(
		f"Executing query: {query} " + f"params: {tuple(kwargs.values())}"
		)
		self.logger.info(f"Executing query: {query} " + f"params: {tuple(kwargs.values())}")
		try:
		# if the db is not open..
		if not self.db.is_open():
		@@ -169,17 +166,13 @@ class CRUDTable(ABC):
		else:
		# raise an exception as we do not know what to do
		raise TypeError(
		"column argument should be of type list or"
		+ f" str not {type(columns)}"
		"column argument should be of type list or" + f" str not {type(columns)}"
		)
		# if the where clause was not set/specified
		if where_clause is None:
		try:
		# make a query to select the columns with no where clause
		query = (
		f"SELECT {select_clause} "
		+ f"FROM {self.schema}.{self.__class__.__name__}"
		)
		query = f"SELECT {select_clause} " + f"FROM {self.schema}.{self.__class__.__name__}"
		# inform the user we are executing the query..
		self.logger.info(f"Executing query: {query}")
		# if the db is not already opened..
		@@ -191,9 +184,7 @@ class CRUDTable(ABC):
		# execute the query
		curr.execute(query)
		except Exception as err:
		self.logger.error(
		"Exception occured when trying to execute " + f"query: {query}"
		)
		self.logger.error("Exception occured when trying to execute " + f"query: {query}")
		self.logger.error(f"Exception Message: {err}")

		# if there is a where clause...
		@@ -206,9 +197,7 @@ class CRUDTable(ABC):
		+ f"{where_clause[0]}"
		)
		# tell the user we are executing their query
		self.logger.info(
		f"Executing query: {query} " + f"params: {where_clause[1]}"
		)
		self.logger.info(f"Executing query: {query} " + f"params: {where_clause[1]}")
		# if the db is not already opened..
		if not self.db.is_open():
		# open the connection to the database
		@@ -315,14 +304,9 @@ class CRUDTable(ABC):
		# construct the where clause with the conversion method
		where_clause = convert_to_where(kwargs)
		# build a delete query with the specified values
		query = (
		f"DELETE FROM {self.schema}.{self.__class__.__name__} "
		+ f"{where_clause[0]}"
		)
		query = f"DELETE FROM {self.schema}.{self.__class__.__name__} " + f"{where_clause[0]}"
		# tell the user that we are executing their query
		self.logger.info(
		f"Executing query: {query}, " + f"params: {where_clause[1]}"
		)
		self.logger.info(f"Executing query: {query}, " + f"params: {where_clause[1]}")
		# if the db is not open..
		if not self.db.is_open():
		# open a connection to the database

src/common/database.py

+1 −2

Original line number	Diff line number	Diff line
		@@ -65,8 +65,7 @@ class Database(ABC):
		DEFAULT_LOG_ENCODING = ce("DATABASE_LOG_ENCODING", "utf-8")
		# define a URI string if URI is perferred to connect
		DEFAULT_URI = (
		f"{DEFAULT_ENGINE}://{DEFAULT_USER}:{str(DEFAULT_PW)}"
		+ f"@{DEFAULT_HOST}/{DEFAULT_DB}"
		f"{DEFAULT_ENGINE}://{DEFAULT_USER}:{str(DEFAULT_PW)}" + f"@{DEFAULT_HOST}/{DEFAULT_DB}"
		)

		def __init__(

src/common/error_codes.py

+2 −3

Original line number	Diff line number	Diff line
		#!/usr/bin/env python3
		# -- coding: utf-8 --
		'''
		"""
		This module defines a data class 'ErrorCodes' that instantiates various error codes used throughout the application.
		It categorizes error codes into distinct sections for database operations, scraping processes,
		templating issues, and provides a default error code for general use. Each error type is associated with specific
		integer values, making it easier to manage and identify errors consistently across different components of the application.
		'''
		"""
		from dataclasses import dataclass


		@dataclass
		class ErrorCodes:

		# Database Errors
		DB_CONNECTION_FAILED: int = 1001
		DB_TIMEOUT: int = 1002

src/common/exceptions.py

+5 −14

Original line number	Diff line number	Diff line
		#!/usr/bin/env python3
		# -- coding: utf-8 --
		'''
		"""
		This module defines custom exception classes for various application components. Each exception class validates
		specific error codes from 'ErrorCodes'. Included are 'ScraperException', 'DatabaseException', 'TemplateException', and 'ParserError',
		each tailored to a specific component and type of error within the application.
		'''
		"""
		from common.error_codes import ErrorCodes


		class CommonException(Exception):
		def __init__(self, message=None, code=None):

		self.code = code if code else ErrorCodes.DEFAULT_ERROR
		self.validate_code(self.code)
		self.message = str(message)
		@@ -31,11 +30,7 @@ class CommonException(Exception):
		class ScraperException(CommonException):
		@staticmethod
		def validate_code(code):
		db_codes = {
		value
		for name, value in vars(ErrorCodes).items()
		if name.startswith("SCRAPE_")
		}
		db_codes = {value for name, value in vars(ErrorCodes).items() if name.startswith("SCRAPE_")}
		if code and code not in db_codes:
		raise ValueError(f"Invalid scraper error code: {code}")

		@@ -43,9 +38,7 @@ class ScraperException(CommonException):
		class DatabaseException(CommonException):
		@staticmethod
		def validate_code(code):
		db_codes = {
		value for name, value in vars(ErrorCodes).items() if name.startswith("DB_")
		}
		db_codes = {value for name, value in vars(ErrorCodes).items() if name.startswith("DB_")}
		if code and code not in db_codes:
		raise ValueError(f"Invalid database error code: {code}")

		@@ -54,9 +47,7 @@ class TemplateException(CommonException):
		@staticmethod
		def validate_code(code):
		db_codes = {
		value
		for name, value in vars(ErrorCodes).items()
		if name.startswith("TEMPLATE_")
		value for name, value in vars(ErrorCodes).items() if name.startswith("TEMPLATE_")
		}
		if code and code not in db_codes:
		raise ValueError(f"Invalid template error code: {code}")