hiplab/data-transport
hiplab
/
data-transport
mirror of https://github.com/lnyemba/data-transport
1
1
Fork 0
Code Issues Packages Projects Releases Wiki Activity

bug fix

This commit is contained in:
Steve Nyemba 2024-02-05 22:35:14 -06:00
parent c3ebd32a40
commit 0f60c748ba
1 changed files with 23 additions and 17 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

40
transport/disk.py
View File

@ -211,18 +211,18 @@ class SQLiteWriter(SQLite,DiskWriter) :
# #
# If the table doesn't exist we should create it # If the table doesn't exist we should create it
# #
def write(self,_data,**_args): # def write(self,_data,**_args):
SQLiteWriter.LOCK.acquire() # SQLiteWriter.LOCK.acquire()
try: # try:
if type(_data) == dict : # if type(_data) == dict :
_data = [_data] # _data = [_data]
_table = self.table if 'table' not in _args else _args['table'] # _table = self.table if 'table' not in _args else _args['table']
_df = pd.DataFrame(_data) # _df = pd.DataFrame(_data)
_df.to_sql(_table,self._engine.connect(),if_exists='append',index=False) # _df.to_sql(_table,self._engine.connect(),if_exists='append',index=False)
except Exception as e: # except Exception as e:
print (e) # print (e)
SQLiteWriter.LOCK.release() # SQLiteWriter.LOCK.release()
def _write(self,info,**_args): def write(self,info,**_args):
""" """
""" """
@ -247,17 +247,23 @@ class SQLiteWriter(SQLite,DiskWriter) :
cursor = self.conn.cursor() cursor = self.conn.cursor()
sql = " " .join(["INSERT INTO ",self.table,"(", ",".join(self.fields) ,")", "values(:values)"]) sql = " " .join(["INSERT INTO ",self.table,"(", ",".join(self.fields) ,")", "values(:values)"])
for row in info : for row in info :
stream =["".join(["",value,""]) if type(value) == str else value for value in row.values()] values = [ str(row[field]) if type(row[field]) not in [list,dict] else json.dumps(row[field],cls=IEncoder) for field in self.fields]
stream = json.dumps(stream,cls=IEncoder) values = ["".join(["'",value,"'"]) for value in values]
stream = stream.replace("[","").replace("]","")
# stream =["".join(["",value,""]) if type(value) == str else value for value in row.values()]
# stream = json.dumps(stream,cls=IEncoder)
# stream = stream.replace("[","").replace("]","")
# print (sql.replace(":values",stream))
self.conn.execute(sql.replace(":values",stream) ) # self.conn.execute(sql.replace(":values",stream) )
self.conn.execute(sql.replace(":values", ",".join(values)) )
# cursor.commit() # cursor.commit()
self.conn.commit() self.conn.commit()
# print (sql) # print (sql)
except Exception as e : except Exception as e :
print ()
print (e) print (e)
pass pass
SQLiteWriter.LOCK.release() SQLiteWriter.LOCK.release()