Managing a store in PostgreSQL

Question

I have been trying to improve python combined with Postgres.

At this moment im only sending two values to database which are etc:

link = "https://www.mathem.se/varor/vegetarisk-fars/vegofars-fairtrade-1000g-anamma"
store = "mathem" 

and I do have few options that I want to be able to do with Postgres and that is to fetch all, commit and count.

And here I do have few options such as to check if the link exists already, if its has been deactivated from before, delete the link, get the ID from the product etc etc:

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

import psycopg2

DATABASE_CONNECTION = {
    "host": "testing.com",
    "database": "test",
    "user": "test",
    "password": "test"
}


def execute_fetch_all(query):
    """
    Fetch all queries
    :param query:
    :return:
    """
    try:
        connection = psycopg2.connect(**DATABASE_CONNECTION)
        cursor = connection.cursor()
        cursor.execute(query)
        response = cursor.fetchall()
        cursor.close()
        connection.close()
        return response

    except (Exception, psycopg2.DatabaseError) as error:
        logger.exception(f"Database error: {error}")
        return


def execute_commit(query):
    """
    Commit database
    :param query:
    :return:
    """
    try:
        connection = psycopg2.connect(**DATABASE_CONNECTION)
        cursor = connection.cursor()
        cursor.execute(query)
        connection.commit()
        cursor.close()
        connection.close()
        return True

    except (Exception, psycopg2.DatabaseError) as error:
        logger.exception(f"Database error: {error}")
        return


def execute_count(query):
    """
    Count elements in database
    :param query:
    :return:
    """
    try:
        connection = psycopg2.connect(**DATABASE_CONNECTION)
        cursor = connection.cursor()
        cursor.execute(query)
        if cursor.rowcount:
            cursor.close()
            connection.close()
            return True
        else:
            cursor.close()
            connection.close()
            return False

    except (Exception, psycopg2.DatabaseError) as error:
        logger.exception(f"Database error: {error}")
        return


def check_if_link_exists(store, link):
    """
    Check if link exists
    :param store:
    :param link:
    :return:
    """
    if execute_count(f"SELECT DISTINCT link FROM public.store_items WHERE store='{store.lower()}' AND link='{link}';"):
        return True
    else:
        return False


def check_if_links_deactivated(store, link):
    """
    Check if link is deactivated
    :param store:
    :param link:
    :return:
    """
    if execute_count(f"SELECT DISTINCT link FROM public.store_items WHERE store='{store.lower()}' AND link='{link}' AND visible='no';"):
        return True
    else:
        return False

def delete_manual_links(store, link):
    """
    Delete given link
    :param store:
    :param link:
    :return:
    """
    execute_commit(f"DELETE FROM public.manual_urls WHERE store='{store.lower()}' AND link='{link}';")
    return True


def get_product_id(store, link):
    """
    Get id from database for specific link
    :param store:
    :param link:
    :return:
    """
    product = execute_fetch_all(f"SELECT DISTINCT id, store, link FROM public.store_items WHERE store='{store.lower()}' AND link='{link}' AND visible='yes';")
    return {"id": product[0][0], "store": product[0][1], "link": product[0][2]}

def get_all_links(store):
    """
    Return all links in database
    :param store:
    :return:
    """
    cur = execute_fetch_all(f"SELECT DISTINCT id, link FROM public.store_items WHERE store='{store.lower()}' AND visible='yes';")
    return [{"id": links[0], "link": links[1]} for links in cur]



def check_if_store_exists(store):
    """
    Check if the store exists in database
    :param store:
    :return:
    """
    if execute_count(f"SELECT DISTINCT store FROM public.store_config WHERE store='{store.lower()}';"):
        return True
    else:
        return False


def register_store(store):
    """
    Register the store
    :param store:
    :return:
    """
    if check_if_store_exists(store=store) is False:
        execute_commit(f"INSERT INTO public.store_config (store) VALUES ('{store.lower()}');")
        return True
    else:
        return False

I wonder if there is a way to maybe even short the code by alot or to also improve when using postgres combined with Python since it is new for me still but I do see potential that I might be able to shorter the code quite alot here

If there is any missing information, please let me know in comments and I will try my best to give the information that I might have forgot to add here

Answer 1

Injection

f"SELECT DISTINCT link FROM public.store_items WHERE store='{store.lower()}' AND link='{link}';"

is wide open to injection attacks. psycopg2 has excellent support for parameters in prepared statements; use that instead of string formatting.

Hard-coded credentials

DATABASE_CONNECTION needs to be externalized to somewhere secure in the operating system environment, for a handful of reasons including protecting the password, and keeping reconfiguration easy. This can take the form of environmental variables or a config file.

Context management

This block:

    connection = psycopg2.connect(**DATABASE_CONNECTION)
    cursor = connection.cursor()
    cursor.execute(query)
    response = cursor.fetchall()
    cursor.close()
    connection.close()

has a number of issues:

• You're opening and closing a new connection every time you do a fetch. Connections are expensive and should be longer-lived.
• You need to replace explicit close with the use of a context-management with statement.
• Your except effectively swallows all exceptions and returns None if there's a failure. That behaviour should not be baked into execute_fetch_all and should exist outside, at the caller level instead. This is true for execute_count and execute_commit as well.

Explicit commit

execute_commit does not benefit from commit() in its current form. Read this for example - https://stackoverflow.com/questions/51880309/what-does-autocommit-mean-in-postgresql-and-psycopg2 - autocommit is the default.

Existence checks

SELECT DISTINCT

is not the best way to check for existence. Instead, consider use of the actual PostgreSQL exists clause.

Dictionary construction

Don't do this:

[{"id": links[0], "link": links[1]} for links in cur]

yourself.

Read https://www.psycopg.org/docs/extras.html#dictionary-like-cursor for the built-in alternative.

Booleans

if x is False:

should be

if not x:

and

if x:
    return True
else:
    return False

should be

return bool(x)

or, if you're comparing a count, more explicitly:

return x > 0