Read created json files inside a given folder

Question

I have been working with where I use watchdogs to be able read whenever I have created a file by the name -> kill_.json and it contains:

*kill_*.json & *start_*.json contains:

{"store": "twitch", "link": "https://www.twitch.tv/xqcow", "id": "2710"}

and currently what my script does is to read the json and the filename. if it contains any of the if elif statement then it will start or kill depending on the filename else we will add it to my database which I dont think its needed to be handed here since its not the code review I wish for... for now :P but if there is any question regarding it I will be answering and adding if needed of course! :)

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

import json
import os
import os.path
import sys
import time

import watchdog.events
import watchdog.observers
from loguru import logger
from watchdog.events import PatternMatchingEventHandler

import lib.database as database


def start_script(payload):
    payload = database.get_product_data(payload["store"], payload["link"])
    logger.info(f'Starting script -> Name: {payload["store"]}_{payload["id"]} | Link: {payload["link"]}')

class DeleteAutomatic(PatternMatchingEventHandler):

    def __enter__(self):
        self.observer = watchdog.observers.Observer()
        self.observer.schedule(
            self,
            path="./flags/"
            recursive=True
        )
        self.observer.start()
        while True:
            time.sleep(1)

    def __exit__(self):
        self.observer.stop()
        self.observer.join()

    def __init__(self):
        super(DeleteAutomatic, self).__init__(
            [
                "*kill_*.json",
                "*start_*.json",
                "*new_url.json"
            ]
        )

    def on_created(self, event):
        while True:
            try:

                with open(event.src_path) as f:
                    payload = json.load(f)

                if "start" in event.src_path:
                    logger.info(f'Starting -> Store: {payload["store"]} | Link: {payload["link"]} | ID: {payload["id"]}')
                    
                elif "kill" in event.src_path:
                    logger.info(f'Deleting -> Store: {payload["store"]} | Link: {payload["link"]} | ID: {payload["id"]}')

                else:
                    for payload in database.get_all_manual_links():

                        logger.info(
                            f'New Manual Url Found -> Store: {payload["store"]} | Link: {payload["link"]} | ID: {payload["id"]}')

                        if not database.link_exists(payload["store"], payload["link"]):
                            database.register_products(payload["store"], product_info)
                            start_script(payload)

                        elif database.links_deactivated(payload["store"], payload["link"]):
                            database.update_products(payload["store"], payload["link"])
                            start_script(payload)

                        else:
                            logger.info(f'Product already monitored: {payload["store"]}_{payload["id"]} | Link: {payload["link"]}')
                            pass

                        database.delete_manual_links(payload["store"], payload["link"])

                break

            except Exception as err:
                logger.debug(f"Error -> {err}")
                time.sleep(1)
                continue

        while True:
            try:
                os.remove(event.src_path)
                break
            except Exception as err:
                logger.debug(f"Error deleting-> {err}")
                time.sleep(1)
                continue

with DeleteAutomatic():
    pass

I wonder if there is a smarter or maybe even cleaner way to improve this code both performance and whatever it can be?

Answer 1

Individual parameters

You're best to replace the payload parameter to start_script with individual store: str, id: str and link: str. Or, if it's common enough for this triple to be carried around, then move it into a @dataclass that has a start_script(self) method.

Method order

This is minor, but typically __init__ should appear as the first method on your class.

Member initialization

    self.observer = watchdog.observers.Observer()
    self.observer.schedule(
        self,
        path="./flags/"
        recursive=True
    )

should be moved to your __init__, since that member has instance lifespan. Linters will tell you that it's not a great idea to initialize new members outside of __init__. The start() can stay in your __enter__.

Standard parameters

The parameters to __exit__ are incorrect, and should be as described in the documentation.

Hanging enter method

Your sleep loop should not be in your __enter__. After start, your enter method needs to yield control to the caller, and the caller needs to have a sleep loop instead.

Parameter-less super()

super(DeleteAutomatic, self)

is Python 2 style; for Python 3 you can omit those parameters.

Start/kill logic

            if "start" in event.src_path:
                logger.info(f'Starting -> Store: {payload["store"]} | Link: {payload["link"]} | ID: {payload["id"]}')
                
            elif "kill" in event.src_path:
                logger.info(f'Deleting -> Store: {payload["store"]} | Link: {payload["link"]} | ID: {payload["id"]}')

should change in a couple of ways: factor out the common logging code to a single statement, with only a Starting / Deleting string varying between the two cases. Also, unpack your payload either to individual variables, or maybe a @dataclass.

One advantage to the @dataclass approach is that this code:

            if "start" in event.src_path:
                logger.info(f'Starting -> Store: {payload["store"]} | Link: {payload["link"]} | ID: {payload["id"]}')
                
            elif "kill" in event.src_path:
                logger.info(f'Deleting -> Store: {payload["store"]} | Link: {payload["link"]} | ID: {payload["id"]}')

            else:
                for payload in database.get_all_manual_links():

                    logger.info(
                        f'New Manual Url Found -> Store: {payload["store"]} | Link: {payload["link"]} | ID: {payload["id"]}')

                    if not database.link_exists(payload["store"], payload["link"]):
                        database.register_products(payload["store"], product_info)
                        start_script(payload)

                    elif database.links_deactivated(payload["store"], payload["link"]):
                        database.update_products(payload["store"], payload["link"])
                        start_script(payload)

                    else:
                        logger.info(f'Product already monitored: {payload["store"]}_{payload["id"]} | Link: {payload["link"]}')
                        pass

                    database.delete_manual_links(payload["store"], payload["link"])

could move to a method on the class; and initializing the class could be a simple Payload(**payload) kwargs.

Retries

        except Exception as err:
            logger.debug(f"Error -> {err}")
            time.sleep(1)
            continue

is risky. You're saying that if any exception occurs, hang for a second and retry. This includes failure modes that you shouldn't reasonably retry on, such as MemoryError. You should have a think about a narrower collection of exceptions that are actually appropriate to retry on.

Syntax

Your code just doesn't run. This syntax is invalid:

        path="./flags/"
        recursive=True

due to a missing comma. I'll give you the benefit of the doubt that that's a copy-and-paste error.

Also, product_info is missing entirely. So your code is incomplete.

Payload references

This:

def start_script(payload):
    payload = database.get_product_data(payload["store"], payload["link"])

is highly suspect. You're passing in payload, using its store and link properties to do a database lookup of what looks to me to be the same payload. Are you sure that's what you want to be doing?

Similarly, this:

            with open(event.src_path) as f:
                payload = json.load(f)

            if "start" in event.src_path:
                logger.info(f'Starting -> Store: {payload["store"]} | Link: {payload["link"]} | ID: {payload["id"]}')
                
            elif "kill" in event.src_path:
                logger.info(f'Deleting -> Store: {payload["store"]} | Link: {payload["link"]} | ID: {payload["id"]}')

            else:
                for payload in database.get_all_manual_links():

ignores the first loaded payload in the third case. So in one out of three conditions, there's actually no point in loading the payload from JSON at all. The logic could be reworked so that the file is only loaded if needed.