I have written small application for inserting data into sqlite. can you please review.
- create data from specified path
- create Table into sqlite database
- 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))
