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 pythonimport argparsefrom datetime import datetimeimport subprocessimport sysimport sqlite3import scandirimport osdef 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 argsclass 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))

1 Answer1

Start asking to get answers

Find the answer to your question by asking.

Explore related questions

See similar questions with these tags.