Whenever I write some code in python that involves a MySQL database, I use this base class that contains all the generic code for the basic CRUD operations. This allows me to focus on writing the queries instead of writing the same boilerplate code every time. It uses the client library PyMySQL and will take care of opening and disposing the connections to the database as well as preventing SQL injection by using parameterized queries. Just wanted to share it so that others can use it too.
Methods
__execute and __openConnection are private methods used internally in the class.
execute is intended for insert, update and delete statements and returns the number of affected rows in the table.
select_all is for select statements and returns a list of the rows that satisfy the query condition(s).
select_one is also for select statements and returns the first row that satisfies the query condition(s).
import pymysql.cursors
class SqlBase(object):
def __init__(self, host, user, password, database):
self.host = host
self.user = user
self.password = password
self.database = database
def execute(self, query:str, parameters:list=[]):
return self.__execute(query, parameters).rowcount
def select_all(self, query:str, parameters:list=[]):
cursor = self.__execute(query, parameters)
return cursor.fetchall()
def select_one(self, query:str, parameters:list=[]):
cursor = self.__execute(query, parameters)
return cursor.fetchone()
def __execute(self, query:str, parameters:list=[]):
try:
with self.__openConnection() as connection:
with connection.cursor() as cursor:
cursor.execute(query, parameters)
connection.commit()
return cursor
except pymysql.Error as e:
print(f"Error while executing query [{query}]: {str(e)}")
def __openConnection(self):
try:
return pymysql.connect(host=self.host, user=self.user, password=self.password, database=self.database, cursorclass=pymysql.cursors.DictCursor)
except pymysql.Error as e:
print(f"Error while opening sql connection: {str(e)}")
Example
Create a class that inherits the SqlBase class and call the base constructor with the database credentials. In this example I have made a user repository with examples of the CRUD operations by using execute, select_one and select_all. Notice how the values are parsed as a separate list of parameters rather than being inserted directly into the query itself. This is very important in order to avoid SQL injection attacks in case the values comes from user input like query parameters.
from sql_base import SqlBase
class UserRepository(SqlBase):
def __init__(self):
SqlBase.__init__(self, "host", "user", "password", "database")
def insert(self, username, email):
query = "INSERT INTO users(username, email) VALUES (%s,%s)"
return self.execute(query, [username, email])
def get(self, username):
query = "SELECT id, username, email FROM users WHERE username=%s"
return self.select_one(query, [username])
def get_all(self):
query = "SELECT id, username, email FROM users"
return self.select_all(query)
def update(self, username, email):
query = "UPDATE users SET email=%s WHERE username=%s"
return self.execute(query, [email, username])
def delete(self, username):
query = "DELETE FROM users WHERE username=%s"
return self.execute(query, [username])