Множество функций, подключенных к базе данных

Сегодня я многому научился и все еще учусь. Я попал в ситуацию, когда понял, что создал базу данных с несколькими функциями, которая выполняет похожие вещи, но с разными запросами. Однако, посмотрев на код, мы можем увидеть аналогичный «путь» этих функций, и я пытался и пытался узнать, как я могу на самом деле реорганизовать дублированный код, если это вообще возможно?

#!/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

Я был бы признателен за всевозможные обзоры, особенно, если есть способ еще больше сократить дублированный код 🙂

Не кодируйте жестко DATABASE_CONNECTION в вашем источнике. Параметры подключения должны находиться вне его, для возможности настройки и безопасности.

Почти каждый ваш комментарий вреднее, чем отсутствие комментариев вообще — это:

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

ложь о том, что класс является функцией, а остальные — это шаблон, который не добавляет ценности.

Ваш образец для select distinct, выбросить все результаты, посчитать строки, выбросить что прочь и свести это к существованию bool нужно изменить. Вместо этого спросите базу данных, существует ли что-нибудь в подзапросе, используя exists пункт. Вместо того, чтобы применять union к black_and_monitored_list, использовать exists(...) or exists(...). Следующее должно работать, хотя я не могу это проверить:

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

Рассмотрите возможность замены вашего data_tuples со словарем, ключи которого соответствуют PsycoPG %(key)s синтаксис для ясности и устойчивости к ошибкам переупорядочения.