Loading src/common/mixins/mssql.py +16 −20 Original line number Diff line number Diff line Loading @@ -6,14 +6,15 @@ import pymssql import traceback class MSSQLMixin(): """Provides a convenient API-like interface to an/a Azure/MS SQL database. """Provides a convenient API-like interface to Microsoft SQL databases. The database object is the interface to the database. :param db_name: the name of the database to connect to """ #user = 'natgas_demo@gasdatafeeddemo' #passwd = '3Ss5$6Uz8#' #host = 'gasdatafeeddemo.database.windows.net' #database = 'gasdatafeed_demo' 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. Loading @@ -33,37 +34,32 @@ class MSSQLMixin(): #connect_timeout=self.connection_info['dbTimeout']) #application_name='ongpy') self.logger.debug('Successfully opened database connection') self.cur = self.connection.cursor() self.cursor = self.connection.cursor() self.logger.debug('Successfully created database cursor') # self.cur.execute("SET search_path TO covidb;") return True except pymssql.OperationalError as e: self.logger.error(f'Database error: {e}') return False return True def close(self): """Close the database connection.""" self.logger.debug('Closing the database coinnection') if hasattr(self, 'cur') and self.cur is not None: self.logger.debug('Closing cursor and setting to None') self.cur.close() self.cur = None if hasattr(self, 'con') and self.con is not None: self.logger.debug('Closing connection and setting to None') """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.logger.debug('Closing sqlalchemy connection') self.engine.dispose() self.logger.debug('Database object successfully closed.') def query(self, query): """Send a query to the database and retrieve the results.""" self.logger.debug(f'Initiating query: {query}') try: self.cur.execute(query) res = self.cur.fetchall() self.cursor.execute(query) res = self.cursor.fetchall() except Exception as e: traceback.print_stack() self.logger.error(f'Problem querying database: {e}') Loading Loading
src/common/mixins/mssql.py +16 −20 Original line number Diff line number Diff line Loading @@ -6,14 +6,15 @@ import pymssql import traceback class MSSQLMixin(): """Provides a convenient API-like interface to an/a Azure/MS SQL database. """Provides a convenient API-like interface to Microsoft SQL databases. The database object is the interface to the database. :param db_name: the name of the database to connect to """ #user = 'natgas_demo@gasdatafeeddemo' #passwd = '3Ss5$6Uz8#' #host = 'gasdatafeeddemo.database.windows.net' #database = 'gasdatafeed_demo' 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. Loading @@ -33,37 +34,32 @@ class MSSQLMixin(): #connect_timeout=self.connection_info['dbTimeout']) #application_name='ongpy') self.logger.debug('Successfully opened database connection') self.cur = self.connection.cursor() self.cursor = self.connection.cursor() self.logger.debug('Successfully created database cursor') # self.cur.execute("SET search_path TO covidb;") return True except pymssql.OperationalError as e: self.logger.error(f'Database error: {e}') return False return True def close(self): """Close the database connection.""" self.logger.debug('Closing the database coinnection') if hasattr(self, 'cur') and self.cur is not None: self.logger.debug('Closing cursor and setting to None') self.cur.close() self.cur = None if hasattr(self, 'con') and self.con is not None: self.logger.debug('Closing connection and setting to None') """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.logger.debug('Closing sqlalchemy connection') self.engine.dispose() self.logger.debug('Database object successfully closed.') def query(self, query): """Send a query to the database and retrieve the results.""" self.logger.debug(f'Initiating query: {query}') try: self.cur.execute(query) res = self.cur.fetchall() self.cursor.execute(query) res = self.cursor.fetchall() except Exception as e: traceback.print_stack() self.logger.error(f'Problem querying database: {e}') Loading
