Раньше я создавал ветку, посвященную аналогичному обзору кода, который у меня был до предыдущего обзора кода — я обновил небольшой совет, полученный из предыдущего ответа, и я понял, что забыл добавить комментарии, а также мог добавить 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 ответ
Пожалуйста, просто удалите это:
"""
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
может означать, что хранилище уже существует — или это может означать, что вставка не удалась и были обновлены нулевые строки!