Contact Book: Sample application to perform CRUD operations using Python, MySQL.
MySQL
Python 3.x
Libraries used :
pymsql , beatutifultable
Installation of Libraries using pip
pip install pymsql
pip install beautifultable
This application performs basic CRUD orations to demonstrate Python end-to-end application.
Note: Update and delete methods were not called from main, you can give a try...
contact.py
config.py
util.py
dboperations.py
contactservice.py
main.py
MySQL
Python 3.x
Libraries used :
pymsql , beatutifultable
Installation of Libraries using pip
pip install pymsql
pip install beautifultable
This application performs basic CRUD orations to demonstrate Python end-to-end application.
Note: Update and delete methods were not called from main, you can give a try...
contact.py
from collections import namedtuple Contact = namedtuple("Contact", ["cid","name", "email", "mobile"])
config.py
mysql = { 'host': 'localhost', 'user': 'root', 'passwd': 'password', 'db': 'counselor_db'}
util.py
import pymysql import config 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 {s}'.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
dboperations.py
import config from util import DbUtil import pymysql from contact import Contact class ContactDbOperations: 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()) return 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 get_all_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) return 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) return 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) @staticmethod def delete_contact(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) @staticmethod def get_list_data(rows): return [Contact(*row) for row in rows]
contactservice.py
from dboperations import ContactDbOperations from contact import Contact from beautifultable import BeautifulTable class ContactService: def __init__(self): self.cs = ContactDbOperations() def add_contact(self): name = input("Name : ") email = input("Email : ") mobile = input("Mobile : ") contact = Contact(0, name, email, mobile) self.cs.add_contact(contact) self.show_all_contacts() def show_all_contacts(self): contacts = self.cs.get_all_contacts() ContactService.paint_data_list(contacts) def search_contact(self): name = input("Enter search name :") contacts = self.cs.search_contact(name) if contacts: ContactService.paint_data_list(contacts) else: print("No data found with search name :{}".format(name)) @staticmethod def paint_data(data): table = BeautifulTable() table.column_headers = ["CID", "NAME", "EMAIL", "MOBILE"] table.append_row([data.cid, data.name, data.email, data.mobile]) print(table) @staticmethod def paint_data_list(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)
main.py
if __name__ == '__main__': cs = ContactService() while True: print("*"*40) print("1. Add Contact 2. View All 3.Search 4. Exit") print("*"*40) choice = int(input("Enter your choice :")) if choice == 1: cs.add_contact() elif choice == 2: cs.show_all_contacts() elif choice == 3: cs.search_contact() elif choice == 4: exit() else: print("Enter only 1 to 4 ......")
