Postgres Shopping Database using Python

Question

So I have just started to work with SQL and I have tested my script and is working very well when it comes to my database.

The database is used for shopping. What I mean by that is that I scrape a website that I love (Flowers etc) and then I collect each item that has been updated on the website and add it to my database if the product has not already been added in the database before (Reason why I have a checkIfExists function)

I also have different get Links, get Keywords functions and the reason I have

getLinks = Is to see if there has been added new page in the website that is not in my DB

getBlackList = In case they re-add same product and I don't want it to be too spammy with my logs then I can blacklist it inside the database.

getAllKeywords = Sometimes I want to filter for specific flowers where I have a function that checks whats the name of the product and then compare if those keywords matches from my database

The code I have written is working as I want and very proud of it and I do feel like I have done some steps abit too much. What I mean by that is that for every function I call cur = self.database.cursor() and cur.close() which maybe is not important or there is a newer way to handle this better. In my eyes I have a feeling I am missing something more.

I would appreciate all kind of help, even if I might have asked wrong question in Code review (its my first time and tried to read the description on how to write good) and as I mentioned before... I appreciate all kind of help! :)

#!/usr/bin/python3 # -*- coding: utf-8 -*- from datetime import datetime import psycopg2 class DataBaseAPI: def __init__(self): self.database = psycopg2.connect(host="test", database="test", user="test", password="test" ) def checkIfExists(self, store, link): try: cur = self.database.cursor() sql = f"SELECT DISTINCT link FROM public.store_items WHERE store='{store.title()}' AND visible='yes' AND link='{link}';" cur.execute(sql) cur.close() except (Exception, psycopg2.DatabaseError) as error: print(error) return False if cur.rowcount: return True else: return False def addProduct(self, product): if self.checkIfExists(store=product["store"], link=product["url"]) is False: link = product["url"], store = product["store"], name = product["name"], image = product["image"] visible = "yes" cur = self.database.cursor() sql = f"INSERT INTO public.store_items (store, name, link, image, visible, added_date) VALUES ('{store}', '{name}', '{link}', '{image}', '{visible}', '{datetime.utcnow().strftime('%Y-%m-%d %H:%M:%S')}');" cur.execute(sql) # # Commit the changes to the database self.database.commit() # Close communication with the PostgreSQL database cur.close() return True else: return False def getAllLinks(self, store): cur = self.database.cursor() sql = f"SELECT DISTINCT link FROM public.store_items WHERE store='{store.title()}' AND visible='yes';" cur.execute(sql) cur.close() return [link[0] for link in cur.fetchall()] def getBlackList(self, store): cur = self.database.cursor() sql = f"SELECT DISTINCT link FROM public.store_items WHERE store='{store.title()}' AND visible='no';" cur.execute(sql) cur.close() return [link[0] for link in cur.fetchall()] def getAllKeywords(self, filtered_or_unfiltered): cur = self.database.cursor() sql = f"SELECT DISTINCT keyword FROM public.keywords WHERE filter_type='{filtered_or_unfiltered}';" cur.execute(sql) cur.close() return [keyword[0] for keyword in cur.fetchall()] 

I think thats pretty much it. Just a simple Add and Get database :)

Answer 1

Before anything, read on SQL parameterization and avoid concatenating or interpolating variables directly in SQL statements. Otherwise you open your application to SQL injection and inefficiencies. The module, psycopg2, has much support for parameterizations as their docs indicate.

Secondly, save cur as another self. attribute and handle cursor and connection closings in a __del__ method. Also public. is the default schema in PostgreSQL and hence can be omitted from table names.

from datetime import datetime import psycopg2 class DataBaseAPI: def __init__(self): self.database = psycopg2.connect(host="test", database="test", user="test", password="test") self.cur = self.database.cursor() def checkIfExists(self, store, link): try: # PREPARED STATEMENT WITH %s PLACEHOLDERS sql = """SELECT DISTINCT link FROM store_items WHERE store=%s AND visible=%s AND link=%s; """ # EXECUTE WITH BINDED PARAMS cur.execute(sql, (store.title(), 'yes', link)) except (Exception, psycopg2.DatabaseError) as error: print(error) return False if cur.rowcount: return True else: return False def addProduct(self, product): if self.checkIfExists(store=product["store"], link=product["url"]) is False: link = product["url"], store = product["store"], name = product["name"], image = product["image"] visible = "yes" # PREPARED STATEMENT WITH %s PLACEHOLDERS sql = """INSERT INTO store_items (store, name, link, image, visible, added_date) VALUES (%s, %s, %s, %s, %s, %s);""" # EXECUTE WITH BINDED PARAMS cur.execute(sql, (store, name, link, image, visible, datetime.utcnow().strftime('%Y-%m-%d %H:%M:%S'))) # Commit the changes to the database self.database.commit() return True else: return False def getAllLinks(self, store): # PREPARED STATEMENT WITH %s PLACEHOLDERS sql = """SELECT DISTINCT link FROM store_items WHERE store=%s AND visible=%s; """ # EXECUTE WITH BINDED PARAMS cur.execute(sql, (store.title(), 'yes')) return [link[0] for link in cur.fetchall()] def getBlackList(self, store): # PREPARED STATEMENT WITH %s PLACEHOLDERS sql = """SELECT DISTINCT link FROM store_items WHERE store=%s AND visible=%s; """ # EXECUTE WITH BINDED PARAMS cur.execute(sql, (store.title(), 'no')) return [link[0] for link in cur.fetchall()] def getAllKeywords(self, filtered_or_unfiltered): # PREPARED STATEMENT WITH %s PLACEHOLDERS sql = """SELECT DISTINCT keyword FROM keywords WHERE filter_type=%s; """ # EXECUTE WITH BINDED PARAMS # (INNER PARENS + COMMA IS NOT A TYPO BUT A ONE-ITEM TUPLE) cur.execute(sql, (filtered_or_unfiltered,)) return [keyword[0] for keyword in cur.fetchall()] def __del__(self): # CLOSE OBJECTS self.cur.close() self.database.close()