Mutilple functions connected to database

Question

Today I have learned alot and still learning. I have came to a situation where I have realized that I have created a database with multiple functions that does similar stuff but different queries. However by looking at the code we can see a similar "path" of these functions and I have tried and tried to know how I can actually refactor the duplicated code if its even possible?

#!/usr/bin/python3
# -*- coding: utf-8 -*-
import re
from datetime import datetime
import psycopg2
import psycopg2.extras
DATABASE_CONNECTION = {
 "host": "xxxxxxxxxx",
 "database": "xxxxxxxxxx",
 "user": "xxxxxxxxxx",
 "password": "xxxxxxxxxx"
}
class QuickConnection:
 """
 Function that connects to the database, waits for the execution and then closes
 """
 def __init__(self):
 self.ps_connection = psycopg2.connect(**DATABASE_CONNECTION)
 self.ps_cursor = self.ps_connection.cursor(cursor_factory=psycopg2.extras.DictCursor)
 self.ps_connection.autocommit = True
 def __enter__(self):
 return self.ps_cursor
 """
 TODO - Print to discord when a error happens
 """
 def __exit__(self, err_type, err_value, traceback):
 if err_type and err_value:
 self.ps_connection.rollback()
 self.ps_cursor.close()
 self.ps_connection.close()
 return False
def link_exists(store, link):
 """
 Check if link exists
 :param store:
 :param link:
 :return:
 """
 sql_query = "SELECT DISTINCT link FROM public.store_items WHERE store=%s AND link=%s;"
 with QuickConnection() as ps_cursor:
 data_tuple = (
 store,
 link
 )
 ps_cursor.execute(sql_query, data_tuple)
 # Should return 1 if it exists
 link_exist = ps_cursor.rowcount
 return bool(link_exist)
def links_deactivated(store, link):
 """
 Check if link is deactivated
 :param store:
 :param link:
 :return:
 """
 sql_query = "SELECT DISTINCT link FROM public.store_items WHERE store=%s AND link=%s AND visible=%s;"
 with QuickConnection() as ps_cursor:
 data_tuple = (
 store,
 link,
 "no"
 )
 ps_cursor.execute(sql_query, data_tuple)
 # Should return 1 if link is deactivated
 has_deactivate = ps_cursor.rowcount
 return bool(has_deactivate)
def register_products(store, product):
 """
 Register a product to database
 :param store:
 :param product:
 :return:
 """
 sql_query = "INSERT INTO public.store_items (store, name, link, image, visible, added_date) VALUES (%s, %s, %s, %s, %s, %s);"
 with QuickConnection() as ps_cursor:
 data_tuple = (
 store,
 product["name"].replace("'", ""),
 re.sub(r'\s+', '', product["link"]),
 re.sub(r'\s+', '', product["image"]),
 "yes",
 datetime.utcnow().strftime("%Y-%m-%d %H:%M:%S.%f")
 )
 ps_cursor.execute(sql_query, data_tuple)
 # Should return 1 if we registered
 has_registered = ps_cursor.rowcount
 return bool(has_registered)
def update_products(store, link):
 """
 Update products value
 :param store:
 :param link:
 :return:
 """
 sql_query = "UPDATE public.store_items SET visible=%s, added_date=%s WHERE store=%s AND link=%s;"
 with QuickConnection() as ps_cursor:
 data_tuple = (
 "yes",
 datetime.utcnow().strftime("%Y-%m-%d %H:%M:%S.%f"),
 store,
 link
 )
 ps_cursor.execute(sql_query, data_tuple)
 # Should return 1 if it has been removed
 has_updated = ps_cursor.rowcount
 return bool(has_updated)
def black_and_monitored_list(store, link):
 """
 Check if the link is already blacklisted or being monitored
 :param store:
 :param link:
 :return:
 """
 sql_query = "SELECT store, link FROM public.manual_urls WHERE link_type=%s AND link=%s AND store=%s union SELECT store, link FROM public.store_items WHERE link=%s AND store=%s"
 with QuickConnection() as ps_cursor:
 data_tuple = (
 "blacklist",
 link,
 store,
 link,
 store
 )
 ps_cursor.execute(sql_query, data_tuple)
 # Should return 1 if it has been removed
 is_flagged = ps_cursor.rowcount
 return bool(is_flagged)
def delete_manual_links(store, link):
 """
 Delete given link
 :param store:
 :param link:
 :return:
 """
 sql_query = "DELETE FROM public.manual_urls WHERE store=%s AND link=%s;"
 with QuickConnection() as ps_cursor:
 data_tuple = (
 store,
 link
 )
 ps_cursor.execute(sql_query, data_tuple)
 # Should return 1 if it has been removed
 has_removed = ps_cursor.rowcount
 return has_removed > 0
 
 
def register_store(store):
 """
 Register the store
 :param store:
 :return:
 """
 if not store_exists(store=store):
 sql_query = "INSERT INTO public.store_config (store) VALUES (%s);"
 with QuickConnection() as ps_cursor:
 data_tuple = (
 store,
 )
 ps_cursor.execute(sql_query, data_tuple)
 # Should have 1 if it has been added
 has_added = ps_cursor.rowcount
 return has_added > 0
 return False

I would appreciate all kind of reviews and specially if there is a way to even shorter the duplicated code as well :)

• Peilonrayz: Why are your functions not methods on QuickConnection?
• PythonNewbie: @Peilonrayz Im not sure if I could answer you the question but the idea was to open the database -> execute -> close. Thats the idea but why its not a method is not what I can answer :(

Answer 1

Don't hard-code DATABASE_CONNECTION in your source. The connection parameters need to live outside of it, for configurability and security.

Nearly every comment that you have is more harmful than having no comment at all - this:

class QuickConnection:
 """
 Function that connects ...
 """

lies about a class being a function, and the others are boilerplate that add no value.

Your pattern to select distinct, throw away all the results, count the rows, throw that away and boil it down to an existence bool needs to change. Instead, ask the database whether anything in a subquery exists using the exists clause. Rather than applying a union to black_and_monitored_list, use exists(...) or exists(...). The following should work, though I can't test it:

select exists (
 select 1 from fruit_urls 
 where link=%(link)s and store=%(store)s and link_type=%(type)s
) or exists(
 select 1 from store_items
 where link=%(link)s and store=%(store)s
);

Consider replacing your data_tuples with a dictionary whose keys correspond to PsycoPG %(key)s syntax, for clarity and robustness against reordering errors.

• PythonNewbie: Hi! 1) Regarding the hard-code. I agree. I will change it into a config file instead.
• Reinderien: Re. 3: it's not "instead of a select"; it needs both, i.e. select exists(select ...)
• Reinderien: re. 4: correct.
• Reinderien: I've edited my answer with an example query.
• PythonNewbie: Hi! I have now done improvement and I think I would need a final review once again! I really appreciate and I hope I have not disappointed you! codereview.stackexchange.com/questions/259255/…