Queries using ORM and PostgreSQL

Question

After yesterday suggestions, I was told... recommended to move over to ORM and I did. I choosed to use PeeWee together with PostgreSQL. Queries for storing and get information from postgreSQL

This is how my currenty code looks like:

import pendulum
from peewee import (
    Model,
    TextField,
    IntegerField,
    DateTimeField, BooleanField, CompositeKey
)
from playhouse.pool import PooledPostgresqlDatabase

from config import configuration

postgres_pool = PooledPostgresqlDatabase(
    configuration.postgresql.database,
    host=configuration.path.database.environment,
    user=configuration.postgresql.user,
    password=configuration.postgresql.password,
    max_connections=100,
    stale_timeout=60,
)


# ------------------------------------------------------------------------------- #
# store_config table
# ------------------------------------------------------------------------------- #
class Stores(Model):
    id = IntegerField(column_name='store_id')
    store = TextField(column_name='store')

    class Meta:
        database = postgres_pool
        db_table = "store_config"
        primary_key = CompositeKey('store')

    @classmethod
    # All stores found in the database
    def all_stores(cls):
        try:
            return cls.select(cls.id, cls.store).order_by(cls.id)
        except Stores.DoesNotExist:
            return None

    @classmethod
    # Insert new row if not in the database
    def getsert(cls, store_name):
        try:
            return cls.get_or_create(store=store_name)
        except Stores.IntegrityError:
            return None


# ------------------------------------------------------------------------------- #
# feed_urls table
# ------------------------------------------------------------------------------- #
class Feeds(Model):
    store = TextField(column_name='store')
    link = TextField(column_name='link')

    class Meta:
        database = postgres_pool
        db_table = "feed_urls"

    @classmethod
    # Get all feeds url that is in given store
    def all_feed_urls(cls, store):
        try:
            return cls.select().where(cls.store.contains(store))
        except Feeds.DoesNotExist:
            return None


# ------------------------------------------------------------------------------- #
# sample_keywords table
# ------------------------------------------------------------------------------- #
class Keywords(Model):
    filter_type = BooleanField(column_name='filter_type')
    keyword = TextField(column_name='keyword')

    class Meta:
        database = postgres_pool
        db_table = "sample_keywords"

    @classmethod
    # Return all keywords stored in the database
    def all_keywords(cls):
        try:
            return cls.select()
        except Keywords.DoesNotExist:
            return None


# ------------------------------------------------------------------------------- #
# store_items table
# ------------------------------------------------------------------------------- #
class Products(Model):
    id = IntegerField(column_name='id')
    store = TextField(column_name='store')
    name = TextField(column_name='name')
    link = TextField(column_name='link')
    image = TextField(column_name='image')
    visible = TextField(column_name='visible')
    added_date = DateTimeField(column_name='added_date')

    class Meta:
        database = postgres_pool
        db_table = "store_items"

    @classmethod
    # Check if given store and link exists in the database
    def store_url_exists(cls, store, link):
        try:
            return cls.select().where((cls.store.contains(store)) & (cls.link.contains(link))).store_url_exists()
        except Products.DoesNotExist:
            return None

    @classmethod
    # Return all urls that contains the store and has the visible set to "yes"
    def active_urls(cls, store):
        try:
            return cls.select().where((cls.store.contains(store)) & (cls.visible == "yes"))
        except Products.DoesNotExist:
            return None

    @classmethod
    # Add product to the database
    def add_product(cls, store, pageData):
        try:
            return cls.insert(
                store=store,
                name=pageData.name,
                link=pageData.link,
                image=pageData.image,
                visible="yes",
                added_date=pendulum.now('Europe/Stockholm').format('YYYY-MM-DD HH:mm:ss.SSSSSS')
            ).execute()
        except Products.DoesNotExist:
            return None

For my eyes it starts to look pretty and I do like the new changes I have done however there is still probably more to be improved and yet there is some stuff that I am concerned about that was mentioned in previous thread.

1. I do have a primary key but that was created straight when I created table in postgreSQL. I did put the Primary key as ID (with auto increment) but I do starting to understand that it is not a good option to do that since there is a chance that a I coud have multiple same URL inserted to the database which could get caught by primary key but I did not. I do believe there should be some improvement here

2. In class Products, I use store_url_exists which basically should be together with add_product and the reason of it is basically that if the URL is not in the database then we should add it and now I have created two queries and I believe it would be better to have it as one but not sure how...?

3. Bad labeling for @classmethod?

I do hope you have seen my improvements and hopefully you do see that those recommendation you have given me is what I do take with me and trying to improve :)

Looking forward!

Answer 1

I am not familiar with that ORM so at this time I will only provide superficial comments.

Suggestions anyway:

Keep all your imports and from ... import on top of the file and group them together.

Keep your data model in a separate Python file, and import it. The point is to better separate the different components of your program. Thus your data model can be imported by other scripts. Larger applications are usually divided in multiple Python files anyway.

Regarding the functions like store_url_exists etc, in my opinion they don't belong to your class. Just keep what is strictly necessary for the functioning of your class. But you could move those functions to another helper file. You import it when you need it.

Restructuring your app this way will declutter the main routine and the code will become more compact and more clear as a result :)

I think you should also start taking the habit of logging events, especially exceptions. It doesn't hurt to have a default exception handler for your application.

For instance you have this function:

def getsert(cls, store_name):
    try:
        return cls.get_or_create(store=store_name)
    except Stores.IntegrityError:
        return None

In case of errors it returns None, but you have zero details about the underlying error. You swallowed it. So you cannot immediately investigate and troubleshoot the issue because you have no trace. At least we hope you will check that the function did not return None.

This is not an event you should be ignoring, instead you should log the details of the error to a file for persistence, and notify the user something did go wrong. When something that bad happens you don't continue. Because the program is going to behave badly and unpredictably.

In my eyes this is a critical error, in fact I would probably not even bother handling IntegrityError within that function, but I would instead let the default exception handler take over and terminate the application gracefully.

If the error is recoverable then you can try to rectify the situation. For example if the connection to the DB was lost because of a network failure, you could reconnect automatically. This is actually expected from a mature application. Restoring a connection is more user-friendly than having to close the program and start it again.

Regarding this function:

def all_feed_urls(cls, store):
    try:
        return cls.select().where(cls.store.contains(store))
    except Feeds.DoesNotExist:
        return None

Have you tested that it triggers DoesNotExist if now rows are found ? DoesNotExist is mentioned in the documentation with regards to the .get method but I am not sure how it applies to .Select. My question is whether .Select would return an iterator of rows that is empty or raise DoesNotExist. I cannot test it on my end but encourage you to do so. Always test edge cases :)