7
\$\begingroup\$

I am from Java background, learning Python. Please review my code and guide me to learn Python properly.

This is a sample code to perform all CRUD operations of contact app.

import pymysql
from beautifultable import BeautifulTable


class Contact:

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


class DbUtil:

    @staticmethod
    def get_connection():

        try:
            con = pymysql.connect("localhost", "root", "password", "counselor_db")
            return con
        except pymysql.DatabaseError as error:
            print("While connecting with database :", error)

    @staticmethod
    def close_connection(conn, cursor):

        try:
            if conn:
                conn.close()
            if cursor:
                cursor.close()
        except pymysql.DatabaseError as error:
            print("While closing connection ", error)


class ContactService:

    def add_contact(self, new_contact):
        try:
            conn = DbUtil.get_connection()
            cursor = conn.cursor()
            cursor.execute("insert into contact(name,email,mobile) values(%s,%s,%s)", (new_contact.name
                                                                                       ,new_contact.email
                                                                                       ,new_contact.mobile))
            conn.commit()
            self.view_contacts()
        except pymysql.DatabaseError as error:
            print("While adding contact to DB :", error)
        finally:
            DbUtil.close_connection(conn, cursor)

    def get_all_records(self):

        try:
            conn = DbUtil.get_connection()
            cursor = conn.cursor()
            cursor.execute("select cid,name,email,mobile from contact")
            rows = cursor.fetchall()
            lst = self.get_list_data(rows)
            return lst
        except pymysql.DatabaseError as error:
            print("While geting data from DB :", error)
        finally:
            DbUtil.close_connection(conn, cursor)

    def search_all_records(self, search_str):

        try:
            conn = DbUtil.get_connection()
            cursor = conn.cursor()
            cursor.execute("select cid,name,email,mobile from contact where name like %s ", ('%' + search_str + '%'))

            rows = cursor.fetchall()
            lst = self.get_list_data(rows);
            return lst
        except pymysql.DatabaseError as error:
            print("While geting data from DB :", error)
        finally:
            DbUtil.close_connection(conn, cursor)

    def get_list_data(self, rows):

        lst = []
        for row in rows:
            cid = row[0]
            name = row[1]
            email = row[2]
            mobile = row[3]
            cont = Contact(name, email, mobile, cid)
            lst.append(cont)
        return lst

    def view_contacts(self):
        lst = self.get_all_records()
        table = BeautifulTable()
        table.column_headers = ["Cid", "Name", "Email", "Mobile"]
        for li in lst:
            table.append_row([li.id, li.name, li.email, li.mobile])
        print(table)

    def view_contact(self, con):

        table = BeautifulTable()
        table.column_headers = ["Cid", "Name", "Email", "Mobile"]
        table.append_row([con.cid, con.name, con.email, con.mobile])
        print(table)

    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)
            row = cursor.fetchone()
            cid = row[0]
            name = row[1]
            email = row[2]
            mobile = row[3]
            cont = Contact(name, email, mobile, cid)
            self.view_contact(cont)
        except pymysql.DatabaseError as error:
            print("While geting data from DB :", error)
        finally:
            DbUtil.close_connection(conn, 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.name,
            contact.email, contact.mobile, contact.cid))
            conn.commit()

        except pymysql.DatabaseError as error:
            print("While updating data of DB :", error)
            conn.rollback()
        finally:
            DbUtil.close_connection(conn, cursor)

    def delete_contact(self, cid):

        try:

            conn = DbUtil.get_connection()
            cursor = conn.cursor()
            cursor.execute("delete from contact where cid = %s ",cid)
            conn.commit()

        except pymysql.DatabaseError as error:
            print("While deleting data from DB :", error)

        finally:
            DbUtil.close_connection(conn, cursor)

    def search_contact(self, search_str):
        lst = self.search_all_records(search_str)
        table = BeautifulTable()
        table.column_headers = ["Cid", "Name", "Email", "Mobile"]
        for li in lst:
            table.append_row([li.cid, li.name, li.email, li.mobile])
        print(table)
\$\endgroup\$
4
\$\begingroup\$

You could make your life a bit easier if Contact had the same order as row, or the other way around. Then you could just use tuple unpacking to create a new Contact:

cont = Contact(*row)

Fortunately this is rather easy, just change your queries:

cursor.execute("select name,email,mobile,cid from contact ...")
cont = Contact(*cursor.fetchone())

The Contact class itself is also more complex than needed. Since you only ever need it to act as an object that holds data (and you don't need to modify an existing Contact, at least not in the code you showed), you can just use a collections.namedtuple:

from collections import namedtuple

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

Making Contact a namedtuple should also allows you to use it like this, since it now behaves exactly like a tuple:

cursor.execute("update contact set name = %s ,email = %s, mobile = %s where cid = %s", contact)

Your get_list_data method can also be greatly simplified using this and a list comprehension:

def get_list_data(self, rows):
    return [Contact(*row) for row in rows]
\$\endgroup\$
  • \$\begingroup\$ Why IDE is warning make all methods as @staticmethod ? \$\endgroup\$ – Lakshman A Mar 13 '18 at 13:20
  • \$\begingroup\$ @LakshmanA Probably because these methods do not depend on any instance variable or method anymore (self is not needed) \$\endgroup\$ – Graipher Mar 13 '18 at 13:21
  • \$\begingroup\$ Instead using namedtuple can i make code better by using class and object \$\endgroup\$ – Lakshman A Mar 13 '18 at 13:21
  • 1
    \$\begingroup\$ @LakshmanA You can, if you need the flexibility offered by a class. If not, a namedtuple is easier. \$\endgroup\$ – Graipher Mar 13 '18 at 13:22
  • \$\begingroup\$ Typical in Java : Dao ( Db operations) , Service (business logic) in Python what is the best way to maintain the source code? What is standard directory structure to organize source code ? \$\endgroup\$ – Lakshman A Mar 13 '18 at 13:24
1
\$\begingroup\$

There are some elements that attract my attention more:

Security

If your application is to be deployed, then it is not good you type the credentials just like in plain within the code. There is a wide literature about this point, and I let you google about it.

Exceptions handling

You should improve the way you are dealing with the exceptions. For example, in get_connection() if an exception is raised, then the return statement will not be executed. To be precise, it will be executed but after the except statement is done, and you will get None as the return value. As you can see, this brings confusion.

The good practice is to keep the try block as light and neat as possible by running only the code that actually is susceptible to raise an exception, nothing more. Anything else, you should delegate it for the following blocks.

So I suggest you to rewrite that method as follows:

@staticmethod
def get_connection():
    try:
        con = pymysql.connect("localhost", "root", "password", "counselor_db")
    except pymysql.DatabaseError as error:
        print("While connecting with database :", error)
        raise
    else:
        return con

This way, the try block does what it is supposed to do: run the code which is prone to troubles. Notice the return statement is executed only if everything was Ok.

Do not forget the remaining exceptions

An other issue with that function is that you are dealing with DatabaseError exception only, but you ignored many other ones. But you have two options to remedy to this problem: either you catch them one by one, or do what follows:

@staticmethod
def get_connection():
    try:
        con = pymysql.connect("localhost", "root", "password", "counselor_db")
    except MySQLError as error:
        print('Exception number: {}, value {!r}'.format(error.args[0], error))
        raise
    else:
        return con

Do not mix EAFP with LBYL

In daily life, it is easier to ask for permission than for forgiveness. In Python, it is the opposite way:

Easier to ask for forgiveness than permission. This common Python coding style assumes the existence of valid keys or attributes and catches exceptions if the assumption proves false. This clean and fast style is characterized by the presence of many try and except statements. The technique contrasts with the LBYL style common to many other languages such as C.

This means it makes sens not to mix the try and if statements as you did in close_connection(). So this would be better:

@staticmethod
def close_connection(conn, cursor):    
    try:
        conn.close()
        cursor.close()
    except pymysql.DatabaseError as error:
        print("While closing connection ", error)
        raise

Apart from the fact, here again, the function misses dealing with other exceptions as previously, it would be better that the try block deals with only one of unit of code which is prone to troubles. In this case, both cursor.close() and conn.close() may raise exceptions. So I would prefer to treat them separately in two different try blocks:

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

You may even create 2 different functions for this purpose, in case you are very meticulous with unit testing.

Elegancy

Not critical but still important: you randomly leave blank lines within the code of the same function. Here are the cases where you can leave blank lines:

Surround top-level function and class definitions with two blank lines.

Method definitions inside a class are surrounded by a single blank line.

Extra blank lines may be used (sparingly) to separate groups of related functions. Blank lines may be omitted between a bunch of related one-liners (e.g. a set of dummy implementations).

Use blank lines in functions, sparingly, to indicate logical sections.

Python accepts the control-L (i.e. ^L) form feed character as whitespace; Many tools treat these characters as page separators, so you may use them to separate pages of related sections of your file. Note, some editors and web-based code viewers may not recognize control-L as a form feed and will show another glyph in its place.

\$\endgroup\$
  • 1
    \$\begingroup\$ Thank so much, I made above said changes ( good practices ). Most of things I am not aware of it. Thank you once again. \$\endgroup\$ – Lakshman A Feb 27 '18 at 10:42
0
\$\begingroup\$

This is a sample code to perform all CRUD operations of contact app. After getting lot of help this is the code I could do. Need help or feedback on further enhancements. Thank you

config.py

 mysql = {'host': 'localhost', 'user': 'root','passwd': 'password', 'db': 'cba'}

contact.py 

from collections import namedtuple

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

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)

    @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)

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())
            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 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]

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

main.py

from contactservice import  ContactService

if __name__ == '__main__':
    cs = ContactService()
    while True:
        print("*"*40)
        print("1.Add Contact 2. View All 3. 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:
            exit()
        else:
            print("Enter only 1 to 3 ......")
\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.