I have written small application for inserting data into sqlite. can you please review.

  1. create data from specified path
  2. create Table into sqlite database
  3. Path format /user/proj/file
#!/usr/bin/env python
import argparse
from datetime import datetime
import subprocess
import sys
import sqlite3
import scandir
import os

def parse_args(argv):
    """Parsing Argument"""
    parser = argparse.ArgumentParser(description="This module update data base",
                                     epilog = "update database for each user")
    parser.add_argument('-mng', '-manager', dest='mng', default=['test'], nargs ="*",
                        help="Please provide manager name")
    parser.add_argument('-path', '-p', dest='path', required=True,
                        help="Please provide path from where data needs to load")
    args = parser.parse_args(argv)
    return args

class Singleton(object):
    create Singleton class
    _instances = {}
    def __new__(class_, *args, **kwargs):
        if class_ not in class_._instances:
            class_._instances[class_] = super(Singleton, class_).__new__(class_, *args, **kwargs)
        return class_._instances[class_]

class DB(Singleton):
    Create DataBae, Table and Insert Data into DataBase
    def __init__(self):
        """Invoke parent class and return one instance"""
        super(DB, self).__init__(self)
        self.db_file = "EmpRun"

    def create_db(self, db_file):
        """create a database file"""
            conn = sqlite3.connect(db_file)
        except Error as e:
            print e

    def create_connection(self):
        create a DataBase connection to SQLlite DataBase
        specified by DataBase connection
        Input : DataBase file
        Output : Connection object or None
            con = sqlite3.connect(self.db_file)
            return con
        except Error as e:
            print (e)
        return None

    def excute_sql(self, conn, sql_query, sql_query_data=""):
        Excute sql query into SQLite DataBase
        Input : connection object
        Ouput : Sql query executation
        Return : Pass or Fail
            c = conn.cursor()
            if sql_query_data:
                c.execute(sql_query, sql_query_data)

        except sqlite3.Error as e:
            print (e)
            return 1
        return 0

    def create_table(self, conn):
        create a table from the create_table_sql statement
        Input : connection object
        Output : create table statement
        return :
        emp_run_detail = """ CREATE TABLE IF NOT EXISTS EMPRUN(
                        emp_name varname(50),
                        proj varname(100),
                        file_data text NOT NULL,
                        time varchar(50) NOT NULL
        if not self.excute_sql(conn, emp_run_detail):
            print "Query : %s Excuted successfully" %(emp_run_detail)
            print "Query : %s Excuted Failed" %(emp_run_detail)

    def extract_info_user(self, index, data, f_name_path):
        Input : Index, data
        Ouput : Tuple with user, proj, file_info
        user = data[index+1]
        proj = data[-2]
        f_name = []
        with open(f_name_path, "r") as fp:
            f_name = fp.read()
        ttime = str(datetime.now())  
        if user == proj:
            proj = user + "_" + os.path.basename(f_name_path) + "_" + ttime
        return(user, proj, f_name, ttime)

    def extract_info(self, path):
        Input : path where all information is stored
        Ouput : return information as list
        file_info = []
        for root, dirs, files in scandir.walk(path):
            for fi in files:
                file_info.append(os.path.join(root, fi))
        user_info = []
        lpath = os.path.basename(path)
        for info in file_info:
            f_data = info.split("/")
            f_index = f_data.index(lpath)
            user_info.append(self.extract_info_user(f_index, f_data, info))
        return user_info

    def insert_data(self, path, conn):
        Insert Data into Table
        Input : Path, where data is located
              : Data Inserted into table
        output : query successful
        emp_run_data =  self.extract_info(path)
        query = """INSERT INTO EMPRUN(emp_name, proj, file_data, time) VALUES(
              ?, ?, ?, ?)"""
        for emp_run in emp_run_data:
            if not self.excute_sql(conn, query, emp_run):
                print "Query : %s Excuted successfully" %(list(emp_run))
                print "Query : %s Excuted Failed" %(list(emp_run))

def main(args):

    database = DB()
    dcon = database.create_connection()
    database.insert_data(args.path, dcon)

if __name__ == "__main__":
    ARGV = parse_args(sys.argv[1:])

1 Answer 1


Here are some of the things I would work on:

  • PEP8 code style violations, to name a few:
    • group and sort imports correctly
    • two blank lines after the import statements
    • docstrings should start with a capital letter, end with a dot
    • two blank lines between the class and function definitions
  • remove unused subprocess import
  • fix typos: excute_sql -> execute_sql, Ouput -> Output, executation -> execution, Excute -> Execute, SQLlite -> SQLite, DataBae -> DataBase
  • use print() as a function for Python 3.x compatibility
  • f_name = [] is unused, remove it
  • I also see Error is not defined here: except Error as e: - did you mean except sqlite3.Error as e:?
  • there is no need to return None at the end of the create_connection() function

I would also re-format the SQL queries for better readability, for instance, replacing:

query = """INSERT INTO EMPRUN(emp_name, proj, file_data, time) VALUES(
      ?, ?, ?, ?)"""


query = """
        (emp_name, proj, file_data, time) 
        (?, ?, ?, ?)"""

The try/finally block inside create_db method is not going to work properly - if sqlite3.connect() fails, conn variable would not be defined - hence, conn.close() in finally will fail with a NameError.

It would also be a good idea to run a static code analysis tool like pylint or flake8 against the code - it would catch most of the code style and PEP8 guide violations.


Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

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