Skip to content

Instantly share code, notes, and snippets.

Embed
What would you like to do?
# Function to close the dashdb connection
def dbclose(self):
try:
retrn = ibm_db.close(self.connection)
return retrn
except Exception as dbcloseerror:
logging.error("dbclose Exception %s"%(dbcloseerror))
return False
# Function to check whether the connection is alive or not
def connectioncheck_handler(self):
try:
logging.info("connection is"+str(active(self.connection)))
dbretry = 0
if (active(self.connection) == False):
while (dbretry<3):
self.connection = ibm_db.connect(self.url ,'' ,'')
if active(self.connection) == True:
dbretry = 3
else:
if dbretry == 2:
raise Exception("db retry Error")
else:
dbretry+=1
logging.info("restarted connection is"+str(active(self.connection)))
except Exception as e:
logging.error("The connectioncheck_handler error is %s"%(e))
# Function to create the Table
def dbCreate(self,tablename,col1,col2,col3,col4,col5):
self.connectioncheck_handler()
try:
create_query = "CREATE TABLE "+tablename+" ("+col1+" INT GENERATED ALWAYS AS IDENTITY (START WITH 1 INCREMENT BY 1 MINVALUE 1 NO MAXVALUE NO CYCLE NO CACHE ORDER),"+col2+" VARCHAR(30) NOT NULL,"+col3+" VARCHAR(30) NOT NULL,"+col4+" VARCHAR(30) NOT NULL,"+col5+" TIMESTAMP NOT NULL, PRIMARY KEY("+col2+","+col3+"))ORGANIZE BY ROW;"
statement = ibm_db.exec_immediate(self.connection, create_query)
ibm_db.free_stmt(statement)
except Exception as e:
logging.error("The dbCreate operation error is %s"%(e))
return False
except:
logging.error("The dbCreate operation error is %s"%(ibm_db.stmt_errormsg()))
return False
return True
# Function to Delete a row from the Table
def dbDelete(self,tablename,conditionColumnName1,conditionColumnValue1,conditionColumnName2,conditionColumnValue2):
self.connectioncheck_handler()
try:
delete_query = "DELETE FROM "+self.DatabaseSchema+"."+tablename+" WHERE "+conditionColumnName1+" = \'"+conditionColumnValue1+"\' AND "+conditionColumnName2+" = \'"+conditionColumnValue2+"\' "
statement = ibm_db.exec_immediate(self.connection, delete_query)
ibm_db.free_stmt(statement)
except Exception as e:
logging.error("The dbDelete operation error is %s"%(e))
return False
except:
logging.error("The dbDelete operation error is %s"%(ibm_db.stmt_errormsg()))
return False
return True
# Function to Fetch the Data from the Table
def dbFetch(self,tablename):
self.connectioncheck_handler()
try:
fetch_query = "SELECT * FROM "+self.DatabaseSchema+"."+tablename+""
statement = ibm_db.exec_immediate(self.connection, fetch_query)
dictionary = ibm_db.fetch_assoc(statement)
data = []
while(dictionary!=False):
data.append(dictionary)
dictionary = ibm_db.fetch_assoc(statement)
ibm_db.free_stmt(statement)
except Exception as e:
logging.error("The dbFetch operation error is %s"%(e))
return False
except:
logging.error("The dbFetch operation error is %s"%(ibm_db.stmt_errormsg()))
return False
return data
# Function to Insert data to the created table
def dbInsert(self,tablename,emailid,password,username,dateofcreation):
self.connectioncheck_handler()
try:
insert_query = "INSERT INTO "+self.DatabaseSchema+"."+tablename+" VALUES (DEFAULT,\'"+emailid+"\',\'"+password+"\',\'"+username+"\',\'"+str(dateofcreation)+"\')"
statement = ibm_db.exec_immediate(self.connection, insert_query)
ibm_db.free_stmt(statement)
except Exception as e:
logging.error("The dbInsert operation error is %s"%(e))
return False
except:
logging.error("The dbInsert operation error is %s"%(ibm_db.stmt_errormsg()))
return False
return True
self.connectioncheck_handler()
try:
update_query = "UPDATE "+tablename+" SET "+columnName+" = \'"+str(updatevalue)+"\' WHERE "+conditionColumnName1+" = \'"+str(conditionColumnValue1)+"\' AND "+conditionColumnName2+" = \'"+str(conditionColumnValue2)+"\'"
statement = ibm_db.exec_immediate(self.connection, update_query)
ibm_db.free_stmt(statement)
except Exception as e:
logging.error("The dbUpdate operation error is %s"%(e))
return False
except:
logging.error("The dbUpdate operation error is %s"%(ibm_db.stmt_errormsg()))
return False
return True
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment