Skip to content

Instantly share code, notes, and snippets.

Embed
What would you like to do?
import sys
import pymysql
import logging
class Database:
"""Database connection class."""
def __init__(self, config):
self.host = config.db_host
self.username = config.db_user
self.password = config.db_password
self.port = config.db_port
self.dbname = config.db_name
self.conn = None
def open_connection(self):
"""Connect to MySQL Database."""
try:
if self.conn is None:
self.conn = pymysql.connect(self.host,
user=self.username,
passwd=self.password,
db=self.dbname,
connect_timeout=5)
except pymysql.MySQLError as e:
logging.error(e)
sys.exit()
finally:
logging.info('Connection opened successfully.')
def run_query(self, query):
"""Execute SQL query."""
try:
self.open_connection()
with self.conn.cursor() as cur:
if 'SELECT' in query:
records = []
cur.execute(query)
result = cur.fetchall()
for row in result:
records.append(row)
cur.close()
return records
else:
result = cur.execute(query)
self.conn.commit()
affected = f"{cur.rowcount} rows affected."
cur.close()
return affected
except pymysql.MySQLError as e:
print(e)
finally:
if self.conn:
self.conn.close()
self.conn = None
logging.info('Database connection closed.')
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
You can’t perform that action at this time.