CRUD operations Python and MySQL

Contact Book APP:

This sample application is to demonstrate CRUD operations using Python and MySQL.

install pymysql from the following link:

        http://pymysql.readthedocs.io/en/latest/user/installation.html

--------------------------------------------------------------------------------------

from collections import  namedtuple
import pymysql
import db.config as config
from beautifultable import BeautifulTable

Contact = namedtuple("Contact", ["cid","name", "email", "mobile"])


class DbUtil:
    @staticmethod
    def get_connection():
        try:
            con = pymysql.connect(config.mysql['host'], config.mysql['user'],config.mysql['passwd'],config.mysql['db'])
        except pymysql.MySQLError as error:
            print('Exception number: {}, value {!r}'.format(error.args[0], error))
            raise
        else:
            return con

    @staticmethod
    def close_connection(conn):
        try:
            conn.close()
        except pymysql.MySQLError as error:
            print("While closing connection ...")
            print('Exception number: {}, value {!r}'.format(error.args[0], error))
            raise

    @staticmethod
    def close_cursor(cursor):
        try:
            cursor.close()
        except pymysql.MySQLError as error:
            print("While closing the cursor ...")
            print('Exception number: {}, value {!r}'.format(error.args[0], error))
            raise


class ContactService:

    def add_contact(self, new_contact):
        try:
            connection = DbUtil.get_connection()
            with connection.cursor() as cursor:
                cursor.execute("insert into contact(name,email,mobile) values(%s,%s,%s)", (new_contact.name
                                                                                           , new_contact.email
                                                                                           , new_contact.mobile))
            connection.commit()

        except pymysql.MySQLError as error:
            print("While inserting Data ...")
            print('Exception number: {}, value {!r}'.format(error.args[0], error))
        finally:
            DbUtil.close_connection(connection)
            DbUtil.close_cursor(cursor)

    def get_contact(self, cid):
        try:

            conn = DbUtil.get_connection()
            cursor = conn.cursor()
            cursor.execute("select  cid, name, email, mobile from contact where cid = %s", cid)
            contact = Contact(*cursor.fetchone())
            self.paint_data(contact)
        except pymysql.DatabaseError as error:
            print("While getting data from DB using id... ")
            print('Exception number: {}, value {!r}'.format(error.args[0], error))
        finally:
            DbUtil.close_connection(conn)
            DbUtil.close_cursor(cursor)

    def update_contact(self, contact):
        try:
            conn = DbUtil.get_connection()
            cursor = conn.cursor()
            cursor.execute("update contact set name = %s ,email = %s, mobile = %s where cid = %s", contact)
            conn.commit()

        except pymysql.DatabaseError as error:
            print("While updating ... ")
            print('Exception number: {}, value {!r}'.format(error.args[0], error))
        finally:
            DbUtil.close_connection(conn)
            DbUtil.close_cursor(cursor)

    def show_contacts(self):
        try:
            connection = DbUtil.get_connection()
            with connection.cursor() as cursor:
                cursor.execute("select cid, name, email, mobile from contact")
                rows = cursor.fetchall()

                contacts = self.get_list_data(rows)

                self.paint_data_list(contacts)
        except Exception as error:
            print("While retrieving data ... ")
            print('Exception number: {}, value {!r}'.format(error.args[0], error))
        finally:
            if connection:
               DbUtil.close_connection(connection)
               DbUtil.close_cursor(cursor)
    def search_contact(self, search_str):
        try:
            connection = DbUtil.get_connection()
            with connection.cursor() as cursor:
                cursor.execute("select cid, name, email, mobile from contact where name like %s ", ('%' + search_str + '%'))
                rows = cursor.fetchall()
                contacts = self.get_list_data(rows)
                print(contacts)
                self.paint_data_list(contacts)
        except Exception as error:
            print("While searching Data ...")
            print('Exception number: {}, value {!r}'.format(error.args[0], error))
        finally:
            DbUtil.close_connection(connection)
            DbUtil.close_cursor(cursor)

    def delete_contact(self, cid):
        try:
            connection = DbUtil.get_connection()
            cursor = connection.cursor()
            cursor.execute("delete from contact where cid = %s ", cid)
            connection.commit()

        except pymysql.DatabaseError as error:
            print("While deleting data ...")
            print('Exception number: {}, value {!r}'.format(error.args[0], error))
        finally:
            DbUtil.close_connection(connection)
            DbUtil.close_cursor(cursor)

    def get_list_data(self, rows):
        return [Contact(*row) for row in rows]

    def paint_data(self, data):
        table = BeautifulTable()
        table.column_headers = ["CID", "NAME", "EMAIL", "MOBILE"]
        table.append_row([data.cid, data.name, data.email, data.mobile])
        print(table)

    def paint_data_list(self, data):
        table = BeautifulTable()
        table.column_headers = ["CID", "NAME", "EMAIL", "MOBILE"]
        for li in data:
            table.append_row([li.cid, li.name, li.email, li.mobile])
        print(table)