python 3.x – Mutilple functions connected to database

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 🙂