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?

• Reinderien: Again - the model for this site doesn't allow for in-place edits. You're free to ask for clarification the way you are in comments, but we don't support updates in your question as a result of feedback.
• PythonNewbie: Ohhhh! Im sorry. I thought I was allowed to update from the answers but ok,good to know :D I have asked in the comments :) @Reinderien

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.

• PythonNewbie: Hello! Hi again as well! I really appreciate the time you are taking to improve my knowledge! I have once again updated my code and adapted it the way you have mentioned in your answer. Individual parameters I was not sure regarding the parameters if I did it correct. I do hope it is something you meant the way I did it? Hanging enter method My guess is to move the while loop at the very bottom instead? Parameter-less super() I assume now its more python 3 correctly? :)
• PythonNewbie: Start/kill logic I think I did it correctly now. I have removed the logging code where I saw that I could re-use it instead at the start_script instead and I have also added some comments where I use a subprocess to kill my process but that is nothing I would like a review on :) (Or atleast not needed)
• PythonNewbie: But I wanted to check with you. am I doing it correctly now the way you answered this question? I need to re-work on the Retries of course but the rest of it? I was most concern for the kwargs
• Reinderien: move the while loop at the very bottom instead - correct.