Not a member of Pastebin yet?
Sign Up,
it unlocks many cool features!
- import mysql.connector
- from contextlib import closing
- import logging
- log = logging.getLogger(__name__)
- class Database:
- """
- Contains code for interfacing with a MySql database
- """
- def __init__(self, user, password, database, host):
- self.user = user
- self.password = password
- self.database = database
- self.host = host
- self._connection = None
- @property
- def connection(self):
- if self._connection is None:
- log.info("Opening new database connection")
- self._connection = mysql.connector.connect(
- user=self.user,
- password=self.password,
- database=self.database,
- host=self.host
- )
- log.info("Connection made")
- if not self._connection.is_connected():
- self._connection.reconnect()
- return self._connection
- def clear_connection(self):
- if self._connection is not None:
- log.info("Closing DB connection")
- self._connection.close()
- self._connection = None
- def close(self):
- self.clear_connection()
- def __enter__(self):
- return self
- def __exit__(self, exc_type, exc_value, traceback):
- self.close()
- def __del__(self):
- # Destructor - note this isn't called if circular references are involved. Prefer using with to relying on this.
- self.close()
- def execute_sql(self, sql, sql_args=(), fetch_all=False, commit=True):
- return self._execute_sql(self.connection, sql, sql_args, fetch_all, commit)
- @staticmethod
- def _execute_sql(cnx, sql, sql_args=(), fetch_all=False, commit=True, fetch_one=False):
- result = None
- with closing(cnx.cursor()) as cursor:
- cursor.execute(sql, sql_args)
- if fetch_all:
- result = cursor.fetchall()
- elif fetch_one:
- result = cursor.fetchone()
- if commit:
- cnx.commit()
- log.debug("Executed statement: %s", cursor.statement)
- return result
- @staticmethod
- def _query_result_to_dict(query_result, keys):
- result = {}
- for i, k in enumerate(keys):
- result[k] = tuple([x[i] for x in query_result])
- return result
Add Comment
Please, Sign In to add comment