Merged develope

version: '3.7'

networks:

  com:

services:

  package:

    build: .

    volumes:

      - ./src/:/tmp/src

      - ./test/:/test/

    env_file:

      - ./envfile_mssql

    command: ["sh", "-c", "/test/mssql_test_server/test-mssql-conn && pytest -v --cov=common /test/test_mssql.py --cov-report term-missing"]

    depends_on:

      - mssql

    networks:

      - com

  mssql:

    build: ./test/mssql_test_server

    environment:

      - ACCEPT_EULA=Y

      - SA_PASSWORD=5nowDog5

      - MSSQL_PID=Developer

    ports:

      - "1433:1433"

    networks:

      - com

DATABASE_HOST=mssql

DATABASE_DB=CommonDB

DATABASE_PORT=1433

DATABASE_USER=sa

DATABASE_PW=5nowDog5

DATABASE_TIMEOUT=60

DATABASE_LOG_FILE=/tmp/test.log

DATABASE_LOG_MODE=w

DATABASE_LOG_ENCODING=utf-8

LOGLEVEL=debug

 No newline at end of file

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

#!/usr/bin/env python3

# -*- coding: utf-8 -*-

"""Allow opening with a pymssql connection."""

import pymssql

import traceback

class MSSQLMixin():

    """Provides a convenient API-like interface to Microsoft SQL databases.

    The database object is the interface to the database.

    """

    def __del__(self):

        """ Close the connection to database.

        Ensure function call is superior to Database i.e:

        class MSSQLDB(MSSQLMixin, Database):"""

        self.close()

    def open(self):

        """Open the connection to the database.

        :return: True if connection established, else false"""

        self.logger.debug('Opening Database Object')

        try:

            msg = "\nConnecting information\n" + \

                  f"Database: {self.connection_info['dbName']}\n" + \

                  f"Host: {self.connection_info['dbHost']}\n" + \

                      f"User: {self.connection_info['dbUser']}\n"

            self.logger.debug(msg)

            self.connection = pymssql.connect(

                database=self.connection_info['dbName'],

                user=self.connection_info['dbUser'],

                password=self.connection_info['dbPassword'],

                host=self.connection_info['dbHost'])

            self.logger.debug('Successfully opened database connection')

            self.cursor = self.connection.cursor()

            self.logger.debug('Successfully created database cursor')

            return True

        except pymssql.OperationalError as e:

            self.logger.error(f'Database error: {e}')

            return False

    def close(self):

        """Close the database connection. No logging. """

        if hasattr(self, 'cursor') and self.cursor is not None:

            self.cursor.close()

            self.cursor = None

        if hasattr(self, 'connnection') and self.connection is not None:

            self.connection.commit()

            self.connection.close()

            self.connection = None

        if hasattr(self, 'engine') and self.engine is not None:

            self.engine.dispose()

    def query(self, query):

        """Send a query to the database and retrieve the results."""

        self.logger.debug(f'Initiating query: {query}')

        try:

            self.cursor.execute(query)

            res = self.cursor.fetchall()

        except Exception as e:

            traceback.print_stack()

            self.logger.error(f'Problem querying database: {e}')

            res = None

        return res

                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}')