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_buildd_id(self,buildd_id):
try:
cur = self.conn.cursor(cursor_factory=psycopg2.extras.DictCursor)
cur.execute("SELECT id,name FROM buildclients WHERE id=%s",(buildd_id,))
res = cur.fetchall()
self.conn.commit()
if (res):
buildd = BuildD(res[0]['id'],res[0]['name'])
cur.close()
return buildd
else:
return None
except psycopg2.Error as e:
self.conn.rollback()
raise Exception("Error retrieving buildd with id:" + str(buildd_id) + ". Database error code: " + str(e.pgcode) + " - Details: " + str(e.pgerror))
return None
def put_buildclient(self,name):
try:
cur = self.conn.cursor(cursor_factory=psycopg2.extras.DictCursor)
cur.execute("INSERT into buildclients(name) VALUES (%s) RETURNING id",(remove_nasties(name),))
res = cur.fetchall()
self.conn.commit()
buildd = BuildD(res[0]['id'],name)
cur.close()
return buildd
except psycopg2.Error as e:
self.conn.rollback()
raise Exception("Error adding buildd:" + name + ". Database error code: " + str(e.pgcode) + " - Details: " + str(e.pgerror))
return None
def get_buildclients(self,page=None):
try:
cur = self.conn.cursor(cursor_factory=psycopg2.extras.DictCursor)
if page:
# CONSTANT
offset = (page -1) * self.limit_high;
cur.execute("SELECT id,name FROM buildclients ORDER BY name LIMIT %s OFFSET %s", (self.limit_high,offset,))
else:
cur.execute("SELECT id,name FROM buildclients ORDER BY name")
res = cur.fetchall()
self.conn.commit()
build_clients = []
for i in res:
build_clients.append(BuildD(i['id'],i['name']))
cur.close()
return build_clients
except psycopg2.Error as e:
self.conn.rollback()
raise Exception("Error retrieving buildd list. Database error code: " + str(e.pgcode) + " - Details: " + str(e.pgerror))
return None