Load data in sqlite

Question

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"
 self.create_db(self.db_file)
 def create_db(self, db_file):
 """create a database file"""
 try:
 conn = sqlite3.connect(db_file)
 except Error as e:
 print e
 finally:
 conn.close()
 def create_connection(self):
 """
 create a DataBase connection to SQLlite DataBase
 specified by DataBase connection
 Input : DataBase file
 Output : Connection object or None
 """
 try:
 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
 """
 try:
 c = conn.cursor()
 if sql_query_data:
 c.execute(sql_query, sql_query_data)
 else:
 c.execute(sql_query)
 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)
 else:
 print "Query : %s Excuted Failed" %(emp_run_detail)
 sys.exit(0)
 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))
 else:
 print "Query : %s Excuted Failed" %(list(emp_run))
 sys.exit(0)
def main(args):
 database = DB()
 dcon = database.create_connection()
 database.create_table(dcon)
 database.insert_data(args.path, dcon)
 dcon.commit()
if __name__ == "__main__":
 ARGV = parse_args(sys.argv[1:])
 sys.exit(main(ARGV))

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(
 ?, ?, ?, ?)"""

with:

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

---

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.