☛ Python CRUD operations with mariaDB in Linux

Installation mariadb in linux server step by step

[root@localhost ~]$ python --version

[root@localhost ~]$ pip --version

[root@localhost ~]$ python -m pip install --upgrade pip

[root@localhost ~]$ pip install mysql-connector-python==8.0.17

[root@localhost ~]$ yum -y install mariadb-devel.x86_64

[root@localhost ~]$ systemctl start mariadb

[root@localhost ~]$ systemctl status mariadb

[root@localhost ~]$ systemctl enable mariadb

[root@localhost ~]$ mysql_secure_installation

[root@localhost ~]$ mysql -p

DBConnectionProvider.py

from mysql import connector
class ConnectionProvider:

	@staticmethod
	def myConnectionProvider():
		conn=connector.connect(user='root',password='root',host='localhost',port=3306,database='mypython')
		return conn
#print(ConnectionProvider.myConnectionProvider())

DBEmployee.py

class Employee:

	def __init__(self,id,name,email,mobile):
		self.id=id
		self.name=name
		self.email=email
		self.mobile=mobile

	
	def __str__(self):
		return "{0}  {1}  {2}  {3}".format(self.id,self.name,self.email,self.mobile)	

DBEmployeeService.py

from abc import ABC,abstractmethod
class EmployeeService(ABC):

	
	#create(save, insert query)
	@abstractmethod
	def save(self,employee):
		pass

	#read(view,select Query)
	@abstractmethod
	def view(self):
		pass

	#delete(delete,delete Query)	
	@abstractmethod
	def delete(self,employeeID):
		pass			

	#update(update, update query)
	@abstractmethod
	def update(self,employee):
		pass

DBEmployeeServiceImpl.py

from DBConnectionProvider import ConnectionProvider;
from DBEmployeeService import EmployeeService
class EmployeeServiceImpl(EmployeeService):

	def save(self,employeeObject):
		try:
			#print(employeeObject)
			conn=ConnectionProvider.myConnectionProvider();
			mycursor=conn.cursor(prepared=True)
			sql="insert into mypython.employee(id,name,email,mobile)values(%s,%s,%s,%s)"
			val=(employeeObject.id,employeeObject.name,employeeObject.email,employeeObject.mobile)
			mycursor.execute(sql,val)
		except Exception as e:
			print(e)
		else:
			conn.commit()
			print(mycursor.rowcount, "record inserted.")

	
	def view(self):
		try:
			conn=ConnectionProvider.myConnectionProvider();
			mycursor=conn.cursor(prepared=True)
			sql="select id,name,email,mobile from employee";
			mycursor.execute(sql)
			rs=mycursor.fetchall()
			for employee in rs:
				print(employee)
		except Exception as e:
			print(e)
		else:
			conn.commit()

	def delete(self,employeeID):
		try:
			conn=ConnectionProvider.myConnectionProvider();
			mycursor=conn.cursor(prepared=True)
			sql="delete from employee where id=%s";
			val=(employeeID,)
			mycursor.execute(sql,val)
		except Exception as e:
			print(e)
		else:
			conn.commit()
			print(mycursor.rowcount, "record deleted.")	

	def update(self,employeeObject):			
		try:
			conn=ConnectionProvider.myConnectionProvider();
			mycursor=conn.cursor(prepared=True)
			sql="update employee set name=%s,email=%s,mobile=%s where id=%s"
			val=(employeeObject.name,employeeObject.email,employeeObject.mobile,employeeObject.id)
			mycursor.execute(sql,val)
		except Exception as e:
			print(e)
		else:
			conn.commit()
			print(mycursor.rowcount, "record update.")			

DBApp.py

from DBEmployeeServiceImpl import EmployeeServiceImpl
from DBEmployee import Employee

obj1=EmployeeServiceImpl()	
employeeObject=Employee(1007,"hitesh","hitesh@gmail.com","7894567899")					
#obj1.save(employeeObject)
obj1.view()
#obj1.delete(1003)
obj1.update(employeeObject)
obj1.view()