Write a database class in python

Question

For a hobby project I'm doing at the moment I want to write an object oriented python program.But one of the first issue I'm running into is creating a class for my (MySQL) database.

I am trying to use as few packages as possible and I want to try to write the database class by using only pymysql.The problem is that there is an overwhelming amount of libraries and a lack of explanation on writing a good database class in python.

Any suggestions and especially examples would be much appreciated

• dizballanze: Please, specify the database you use.
• Nils Werner: Not an answer but an alternative: Learn how to master one of the big ORMs like sqlalchemy instead. Writing your own X seems like a nice way to practice (sometimes it is) but in the end you may miss important things or grow tired of the project. If you try and understand/master something well designed and already in existence you will stumble across so many edge cases and intricacies that you aren't even thinking of right now.
• Coen van Campenhout: I actually tried peewee but wasn't too fond of it, is sqlalchemy a better alternative?

Answer 1

When having the exact same situation, I found the mysql-connector-python class with which I created a small "model" to call from other classes. This is a cut-down version which shows various db calls. As you can see I have a config class which holds all db authentication info and more).

# dependancy: mysql-connector-python (https://dev.mysql.com/downloads/connector/python/2.1.html)
import mysql.connector
import time
import config
import HTMLParser
import StringIO
html_parser = HTMLParser.HTMLParser()
try:
 connection = mysql.connector.connect( user=config.DB_USER, password=config.DB_PASSWORD,
 host = config.DB_HOST, database=config.DB_DATABASE, unix_socket=config.UNIX_SOCKET)
 cursor = connection.cursor()
except mysql.connector.Error as err:
 logger.log('Database connection failed for '+config.DB_USER+'@'+config.DB_HOST+'/'+config.DB_DATABASE)
 exit()
def get_bad_words():
 sql = ("SELECT word FROM word_blacklist")
 results = execute(sql)
 return results
def get_moderation_method():
 sql = ("SELECT var_value FROM settings "
 "WHERE var_key = %(key)s")
 results = execute(sql, True, {'key':'moderation_method'})
 return results[0]
def current_events():
 sql = ("SELECT count(id) FROM events WHERE event_date >= DATE_SUB(NOW(), INTERVAL 2 hour) AND event_date <= DATE_ADD(NOW(), INTERVAL 5 hour)")
 results = execute(sql, True)
 return results[0]
def insert_social_post(channel, filter_type, post_id, validate, user_name, user_id, user_profile_picture, text, post_date, image_url, state):
 try:
 san_user_name = html_parser.unescape(user_name.encode('utf-8').strip()).decode("utf8").encode('ascii','ignore')
 except:
 san_user_name = html_parser.unescape(user_name.strip())
 try:
 san_text = html_parser.unescape(text.encode('utf-8').strip()).decode("utf8").encode('ascii','ignore')
 except:
 san_text = html_parser.unescape(text.strip())
 insert_post = ("INSERT IGNORE INTO social_posts "
 "(channel, filter_type, post_id, validate, user_name, user_id, user_profile_picture, text, post_date, image_url, state)"
 "VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)")
 execute(insert_post, False, [channel, filter_type, str(post_id), validate,
 san_user_name.strip(), user_id, user_profile_picture, san_text.strip(), post_date, image_url, state], True)
def delete_posts(ids):
 fmt = ','.join(['%s'] * len(ids))
 cursor.execute("DELETE FROM `social_posts` WHERE id IN (%s)" % fmt,
 tuple(ids))
 connection.commit()
def update_campaigns(campaigns):
 sql = ("UPDATE social_campaigns "
 "SET last_updated = NOW()"
 "WHERE id IN ("+(','.join(str(c) for c in campaigns))+")")
 execute(sql, False, None, True)
def execute(tuple, single = False, args = {}, commit = False):
 cursor.execute(tuple, args)
 if commit == True:
 connection.commit()
 else:
 if single == True:
 return cursor.fetchone()
 else:
 return cursor.fetchall()
def lastrowid():
 return cursor.lastrowid
def close():
 connection.close()

Call upon the class like this:

import db
bad_words = db.get_bad_words()

• NETFLOX: tank you for this example class!

Answer 2

import sqlite3
"""singleton class to deal with db"""
'''same can be use for pymysql just replace the sqlite3 with pymysql'''
class DBConnection:
 instance = None
 def __new__(cls, *args, **kwargs):
 if cls.instance is None:
 cls.instance = super().__new__(DBConnection)
 return cls.instance
 return cls.instance
 def __init__(self, db_name='you-db-name'):
 self.name = db_name
 # connect takes url, dbname, user-id, password
 self.conn = self.connect(db_name)
 self.cursor = self.conn.cursor()
 def connect(self):
 try:
 return sqlite3.connect(self.name)
 except sqlite3.Error as e:
 pass
 def __del__(self):
 self.cursor.close()
 self.conn.close()
 # write your function here for CRUD operations