Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def get_suitearch_id(self,suitearch_id):
try:
cur = self.conn.cursor(cursor_factory=psycopg2.extras.DictCursor)
cur.execute("SELECT id,suite_id,arch_id,master_weight FROM suitearches WHERE id=%s",(suitearch_id,))
res = cur.fetchall()
self.conn.commit()
suitearch = SuiteArch(res[0]['id'],self.get_suite_id(res[0]['suite_id']),self.get_arch_id(res[0]['arch_id']),res[0]['master_weight'])
cur.close()
return suitearch
except psycopg2.Error as e:
self.conn.rollback()
raise Exception("Error retrieving suite arch with id:" + str(suitearch_id) + ". Database error code: " + str(e.pgcode) + " - Details: " + str(e.pgerror))
return None
def get_suitearches(self):
try:
cur = self.conn.cursor(cursor_factory=psycopg2.extras.DictCursor)
cur.execute("SELECT id,suite_id,arch_id,master_weight FROM suitearches ORDER BY master_weight DESC")
res = cur.fetchall()
self.conn.commit()
suite_arches = []
for i in res:
suite_arches.append(SuiteArch(i['id'],self.get_suite_id(i['suite_id']),self.get_arch_id(i['arch_id']),i['master_weight']))
cur.close()
return suite_arches
except psycopg2.Error as e:
self.conn.rollback()
raise Exception("Error retrieving suite arches list. Database error code: " + str(e.pgcode) + " - Details: " + str(e.pgerror))
return None
def get_suitearch_by_suite_name(self,suite,arch):
try:
cur = self.conn.cursor(cursor_factory=psycopg2.extras.DictCursor)
cur.execute("SELECT id,suite_id,arch_id,master_weight FROM suitearches WHERE suite.id=%s, arch.id=%s",(suite.id,arch.id))
res = cur.fetchall()
self.conn.commit()
suitearch = SuiteArch(res[0]['id'],self.get_suite_id(res[0]['suite_id']),self.get_arch_id(res[0]['arch_id']),res[0]['master_weight'])
cur.close()
return suitearch
except psycopg2.Error as e:
self.conn.rollback()
raise Exception("Error retrieving suite arch with suite and arch:. Database error code: " + str(e.pgcode) + " - Details: " + str(e.pgerror))
return None
def put_suitearch(self,suite_id,arch_id,master_weight = 0):
try:
cur = self.conn.cursor(cursor_factory=psycopg2.extras.DictCursor)
cur.execute("INSERT into suitearches(suite_id,arch_id,master_weight) VALUES (%s, %s, %s) RETURNING id",(remove_nasties(suite_id),remove_nasties(arch_id),remove_nasties(master_weight)))
res = cur.fetchall()
self.conn.commit()
suitearch = SuiteArch(res[0]['id'],self.get_suite_id(suite_id),self.get_arch_id(arch_id),master_weight)
cur.close()
return suitearch
except psycopg2.Error as e:
self.conn.rollback()
raise Exception("Error adding suite arch:" + suite_id + arch_id + ". Database error code: " + str(e.pgcode) + " - Details: " + str(e.pgerror))
return None