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()