Несколько функций, подключенных к SQL

Раньше я создавал ветку, посвященную аналогичному обзору кода, который у меня был до предыдущего обзора кода — я обновил небольшой совет, полученный из предыдущего ответа, и я понял, что забыл добавить комментарии, а также мог добавить exists(SELECT 1...). Однако я заменил код, в котором я использую dict вместо кортежей 🙂

Основная причина написания этого SQL — это в основном новые знания, а также сделать мой сценарий более полезным, например, я могу повторно использовать эти функции в нескольких сценариях. Где я могу легко поддерживать один сценарий вместо того, чтобы использовать один и тот же код в нескольких сценариях.

Меня больше всего беспокоит то, что я действительно вижу небольшой потенциал рефакторинга, когда я, скорее всего, мог бы еще больше укоротить код, но вот я застрял на нем и считаю, что может быть небольшой шанс, что это невозможно, но я бы с удовольствием еще раз присмотреть за ним OO

#!/usr/bin/python3
# -*- coding: utf-8 -*-
from datetime import datetime

import psycopg2
import psycopg2.extras

from config import configuration

DATABASE_CONNECTION = {
    "host": configuration.path.database.environment,
    "database": configuration.postgresql.database,
    "user": configuration.postgresql.user,
    "password": configuration.postgresql.password
}


class QuickConnection:
    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:
    """

    dict_tuple = {"store": store, "link": link}

    sql_query = "SELECT EXISTS (SELECT 1 FROM public.store_items WHERE store=%(store)s AND link=%(link)s);"

    with QuickConnection() as ps_cursor:
        ps_cursor.execute(sql_query, dict_tuple)
        return ps_cursor.fetchone()[0]


def register_products(store, product):
    """
    Register a product to database
    :param store:
    :param product:
    :return:
    """

    dict_tuple = {"store": store, "name": product["name"], "link": product["link"], "image": product["image"], "visible": "yes", "added_date": datetime.utcnow().strftime("%Y-%m-%d %H:%M:%S.%f")}

    sql_query = "INSERT INTO public.store_items (store, name, link, image, visible, added_date) VALUES (%(store)s, %(name)s, %(link)s, %(image)s, %(visible)s, %(added_date)s);"

    with QuickConnection() as ps_cursor:
        ps_cursor.execute(sql_query, dict_tuple)
        return bool(ps_cursor.rowcount)


def update_products(store, link):
    """
    Update products value
    :param store:
    :param link:
    :return:
    """

    dict_tuple = {"store": store, "link": link, "visible": "yes", "added_date": datetime.utcnow().strftime("%Y-%m-%d %H:%M:%S.%f")}

    sql_query = "UPDATE public.store_items SET visible=%(visible)s, added_date=%(added_date)s WHERE store=%(store)s AND link=%(link)s;"

    with QuickConnection() as ps_cursor:
        ps_cursor.execute(sql_query, dict_tuple)
        return bool(ps_cursor.rowcount)


def black_and_monitored_list(store, link):
    """
    Check if the link is already blacklisted or being monitored
    :param store:
    :param link:
    :return:
    """

    dict_tuple = {"type": "blacklist", "link": link, "store": store}

    sql_query = "SELECT EXISTS (SELECT 1 FROM manual_urls WHERE link=%(link)s AND store=%(store)s AND link_type=%(type)s) OR EXISTS (SELECT store, link FROM store_items WHERE link=%(link)s AND store=%(store)s);"

    with QuickConnection() as ps_cursor:
        ps_cursor.execute(sql_query, dict_tuple)
        return ps_cursor.fetchone()[0]


def delete_manual_links(store, link):
    """
    Delete given link
    :param store:
    :param link:
    :return:
    """

    dict_tuple = {"store": store, "link": link}

    sql_query = "DELETE FROM public.manual_urls WHERE store=%(store)s AND link=%(link)s;"

    with QuickConnection() as ps_cursor:
        ps_cursor.execute(sql_query, dict_tuple)
        return bool(ps_cursor.rowcount)


def get_product_data(store, link):
    """
    Get id from database for specific link
    :param store:
    :param link:
    :return:
    """

    dict_tuple = {"store": store, "link": link, "visible": "yes"}

    sql_query = "SELECT id, store, link FROM public.store_items WHERE store=%(store)s AND link=%(link)s AND visible=%(visible)s;"

    with QuickConnection() as ps_cursor:
        ps_cursor.execute(sql_query, dict_tuple)
        product = ps_cursor.fetchone()

        return {"id": product["id"], "store": product["store"], "link": product["link"]}


def get_all_keywords(positive_or_negative):
    """
    Get all keywords
    :param positive_or_negative:
    :return:
    """

    dict_tuple = {"keyword": positive_or_negative}

    sql_query = "SELECT keyword FROM public.keywords WHERE filter_type = %(keyword)s;"

    with QuickConnection() as ps_cursor:
        ps_cursor.execute(sql_query, dict_tuple)
        return [keyword["keyword"] for keyword in ps_cursor]



def store_exists(store):
    """
    Check if the store exists in database
    :param store:
    :return:
    """

    dict_tuple = {"store": store}

    sql_query = "SELECT EXISTS (SELECT 1 FROM public.store_config WHERE store = %(store)s);"

    with QuickConnection() as ps_cursor:
        ps_cursor.execute(sql_query, dict_tuple)
        return ps_cursor.fetchone()[0]


def register_store(store):
    """
    Register the store
    :param store:
    :return:
    """
    if not store_exists(store=store):

        dict_tuple = {"store", store}

        sql_query = "INSERT INTO public.store_config (store) VALUES (%(store)s);"

        with QuickConnection() as ps_cursor:
            ps_cursor.execute(sql_query, dict_tuple)
            return bool(ps_cursor.rowcount)

    return False

1 ответ
1

Пожалуйста, просто удалите это:

"""
Check if link exists
:param store:
:param link:
:return:
"""

Я предполагаю, что ваша IDE создает этот шаблон для вас, ожидая, что вы напишете содержательную документацию для параметров и возвращаемого значения. Вы еще не сделали этого, и методы достаточно очевидны, поэтому вам следует полностью отказаться от строк документации. Что бы это ни стоило, я нахожу добавление подсказок типа PEP484 — т.е. def link_exists(store: str, link: str) -> bool (предположительно) — в любом случае, чтобы быть информативнее.

Подумайте о замене этого:

    return ps_cursor.fetchone()[0]

с участием

exists, = ps_cursor.fetchone()
return exists

так как он бросит, если вы неожиданно получите более одного результата.

dict_tuple не подходящее имя для этой переменной — это не кортеж; это просто изречение. В любом случае, более информативное имя было бы примерно таким: query_params.

Ваш visible столбец подозрительный; это похоже на строку, принимающую yes. Вы должны переписать его как логическое значение в базе данных.

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

register_store имеет загадочное возвращаемое значение. False может означать, что хранилище уже существует — или это может означать, что вставка не удалась и были обновлены нулевые строки!

    Добавить комментарий

    Ваш адрес email не будет опубликован. Обязательные поля помечены *