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_packages(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,version,name FROM package ORDER BY name,id LIMIT %s OFFSET %s", (self.limit_high,offset,))
else:
cur.execute("SELECT id,version,name FROM package ORDER BY name,id")
res = cur.fetchall()
self.conn.commit()
packages = []
for i in res:
packages.append(Package(i['id'],i['version'],i['name']))
cur.close()
return packages
except psycopg2.Error as e:
self.conn.rollback()
raise Exception("Error retrieving packages list. Database error code: " + str(e.pgcode) + " - Details: " + str(e.pgerror))
return None
def get_package_id(self,package_id):
try:
cur = self.conn.cursor(cursor_factory=psycopg2.extras.DictCursor)
cur.execute("SELECT id,version,name FROM package WHERE id=%s",(package_id,))
res = cur.fetchall()
self.conn.commit()
package = Package(res[0]['id'],res[0]['version'],res[0]['name'])
cur.close()
return package
except psycopg2.Error as e:
self.conn.rollback()
raise Exception("Error retrieving package with id:" + str(package_id) + ". Database error code: " + str(e.pgcode) + " - Details: " + str(e.pgerror))
return None
def get_packagenames(self):
try:
cur = self.conn.cursor(cursor_factory=psycopg2.extras.DictCursor)
cur.execute("SELECT DISTINCT (name), name FROM package GROUP BY name ORDER BY name") # We only care about a unique list of names
res = cur.fetchall()
self.conn.commit()
packages = []
for i in res:
packages.append(Package(None,None,i['name'])) # TODO: these may actually be useful to have still.
cur.close()
return packages
except psycopg2.Error as e:
self.conn.rollback()
raise Exception("Error retrieving packages list. Database error code: " + str(e.pgcode) + " - Details: " + str(e.pgerror))
return None
def get_package_byvalues(self,name,version):
try:
cur = self.conn.cursor(cursor_factory=psycopg2.extras.DictCursor)
cur.execute("SELECT id,name,version FROM package WHERE name=%s AND version=%s",(name,version))
res = cur.fetchall()
self.conn.commit()
packages = []
for i in res:
packages.append(Package(i['id'],i['version'],i['name']))
cur.close()
return packages
except psycopg2.Error as e:
self.conn.rollback()
raise Exception("Error retrieving package by values:" + name + version + ". Database error code: " + str(e.pgcode) + " - Details: " + str(e.pgerror))
return None
def get_packages_byname(self, name):
try:
cur = self.conn.cursor(cursor_factory=psycopg2.extras.DictCursor)
cur.execute("SELECT id,version,name FROM package WHERE name=%s",(name,))
res = cur.fetchall()
self.conn.commit()
packages = []
for i in res:
packages.append(Package(i['id'],i['version'],i['name']))
cur.close()
return packages
except psycopg2.Error as e:
self.conn.rollback()
raise Exception("Error retrieving package with name:" + str(name) + ". Database error code: " + str(e.pgcode) + " - Details: " + str(e.pgerror))
return None
def put_package(self,version,name):
try:
cur = self.conn.cursor(cursor_factory=psycopg2.extras.DictCursor)
cur.execute("INSERT into package(version,name) VALUES (%s, %s) RETURNING id",(remove_nasties(version),remove_nasties(name)))
res = cur.fetchall()
self.conn.commit()
package = Package(res[0]['id'],version,name)
cur.close()
return package
except psycopg2.Error as e:
self.conn.rollback()
raise Exception("Error adding package:" + name + version + ". Database error code: " + str(e.pgcode) + " - Details: " + str(e.pgerror))
return None