background_tasks.py

Go to the documentation of this file.

r"""
# +==== BEGIN CatFeeder =================+
# LOGO:
# ..............(..../\
# ...............)..(.')
# ..............(../..)
# ...............\‍(__)|
# Inspired by Joan Stark
# source https://www.asciiart.eu/
# animals/cats
# /STOP
# PROJECT: CatFeeder
# FILE: background_tasks.py
# CREATION DATE: 11-10-2025
# LAST Modified: 14:43:26 19-12-2025
# DESCRIPTION:
# This is the backend server in charge of making the actual website work.
# /STOP
# COPYRIGHT: (c) Cat Feeder
# PURPOSE: File in charge of setting up the cron jobs for the server.
# // AR
# +==== END CatFeeder =================+
"""
 
 
from typing import Union, Any, Dict, Tuple, Callable
from apscheduler.job import Job
from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.schedulers import SchedulerAlreadyRunningError, SchedulerNotRunningError
from display_tty import Disp, initialise_logger
from ..core import FinalClass
 
 
class BackgroundTasks(metaclass=FinalClass):
    """_summary_
        This is the class that is in charge of scheduling background tasks that need to run on intervals
    """
 
    disp: Disp = initialise_logger(__qualname__, False)
 
    def __init__(self, success: int = 0, error: int = 84, debug: bool = False) -> None:
        # ------------------------ The logging function ------------------------
        self.disp.update_disp_debug(debug)
        self.disp.log_debug("Initialising...")
        # -------------------------- Inherited values --------------------------
        self.success: int = success
        self.error: int = error
        self.debug: bool = debug
        # ------------------------ The scheduler class  ------------------------
        self.scheduler = BackgroundScheduler()
        self.disp.log_debug("Initialised")
 
    def __del__(self) -> None:
        """_summary_
            The destructor of the class
        """
        self.disp.log_info("Stopping background tasks.", "__del__")
        exit_code = self.safe_stop()
        msg = f"The cron exited with status {exit_code}."
        if exit_code != self.success:
            self.disp.log_error(msg, "__del__")
        else:
            self.disp.log_debug(msg, "__del__")
 
    def safe_add_task(self, func: Callable, args: Union[Tuple, None] = None, kwargs: Union[Dict, None] = None, trigger: Union[str, Any] = "interval", seconds: int = 5) -> Union[int, Job]:
        """_summary_
            A non-crashing implementation of the add_task function.
 
        Args:
            func (Callable): _description_: The function to be called when it is time to run the job
            args (Union[Tuple, None], optional): _description_. Defaults to None.: Arguments you wish to pass to the function when executed.
            kwargs (Union[Dict, None], optional): _description_. Defaults to None.: Arguments you wish to pass to the function when executed.
            trigger (Union[str, Any], optional): _description_. Defaults to "interval".
            seconds (int, optional): _description_. Defaults to 5. The amount of seconds to wait before executing the task again (I don't think it is effective for the cron option)
 
        Returns:
            Union[int, Job]: _description_: returns self.error if there was an error, otherwise, returns a Job instance.
        """
        try:
            data = self.add_task(
                func=func,
                args=args,
                kwargs=kwargs,
                trigger=trigger,
                seconds=seconds
            )
            if data is None:
                self.disp.log_error(
                    "The add_task function returned None, which is unexpected.",
                    "safe_add_task"
                )
                return self.error
            return data
        except ValueError as e:
            self.disp.log_error(
                f"Runtime Error for add_task. {e}",
                "safe_add_task"
            )
            return self.error
 
    def safe_start(self) -> int:
        """_summary_
            This function is in charge of starting the scheduler. In a non-breaking way.
 
        Returns:
            int: _description_: Will return self.success if it worked, otherwise self.error.
        """
        try:
            data = self.start()
            if data is None:
                self.disp.log_error(
                    "The start function returned None, which is unexpected.",
                    "safe_start"
                )
                return self.error
            return data
        except RuntimeError as e:
            self.disp.log_error(
                f"Runtime Error for start. {e}",
                "safe_start"
            )
            return self.error
 
    def safe_pause(self, pause: bool = True) -> int:
        """_summary_
            This function is in charge of pausing the scheduler. In a non-breaking way.
 
        Args:
            pause (bool, optional): _description_: This is the boolean that will determine if the scheduler should be paused or not. Defaults to True.
 
        Returns:
            int: _description_: Will return self.success if it worked, otherwise self.error
        """
        try:
            data = self.pause(pause=pause)
            if data is None:
                self.disp.log_error(
                    "The pause function returned None, which is unexpected.",
                    "safe_pause"
                )
                return self.error
            return data
        except RuntimeError as e:
            self.disp.log_error(
                f"Runtime Error for start. {e}",
                "safe_pause"
            )
            return self.error
 
    def safe_resume(self) -> int:
        """_summary_
            This function is in charge of resuming the scheduler. In a non-breaking way.
 
        Returns:
            int: _description_: Will return self.success if it worked, otherwise self.error.
        """
        try:
            data = self.resume()
            if data is None:
                self.disp.log_error(
                    "The resume function returned None, which is unexpected.",
                    "safe_resume"
                )
                return self.error
            return data
        except RuntimeError as e:
            self.disp.log_error(
                f"Runtime Error for start. {e}",
                "safe_resume"
            )
            return self.error
 
    def safe_stop(self, wait: bool = True) -> int:
        """_summary_
            This function is in charge of stopping the scheduler. In a non-breaking way.
 
        Args:
            wait (bool, optional): _description_: Wait for the running tasks to finish. Defaults to True.
 
        Returns:
            int: _description_: will return self.success if it succeeds, otherwise self.error
        """
        try:
            data = self.stop(wait=wait)
            if data is None:
                self.disp.log_error(
                    "The stop function returned None, which is unexpected.",
                    "safe_stop"
                )
                return self.error
            return data
        except RuntimeError as e:
            self.disp.log_error(
                f"Runtime Error for start. {e}",
                "safe_stop"
            )
            return self.error
 
    def _to_dict(self, data: Union[Any, None] = None) -> dict:
        """_summary_
            Convert any data input into a dictionnary.
        Args:
            data (Union[Any, None], optional): _description_. Defaults to None. This is the data you are providing.
 
        Returns:
            dict: _description_: A dictionnary crea ted with what could be extracted from the data.
        """
        if data is None:
            return {"none": None}
        if isinstance(data, dict) is True:
            return data
        if isinstance(data, (list, tuple)) is True:
            res = {}
            for i in list(data):
                res[i] = None
            return res
        return {"data": data}
 
    def add_task(self, func: Callable, args: Union[Tuple, None] = None, kwargs: Union[Dict, None] = None, trigger: Union[str, Any] = "interval",  seconds: int = 5) -> Union[Job, None]:
        """_summary_
            Function in charge of adding an automated call to functions that are meant to run in the background.
            They are meant to run on interval.
 
        Args:
            func (Callable): _description_: The function to be called when it is time to run the job
            args (Union[Tuple, None], optional): _description_. Defaults to None.: Arguments you wish to pass to the function when executed.
            kwargs (Union[Dict, None], optional): _description_. Defaults to None.: Arguments you wish to pass to the function when executed.
            trigger (Union[str, Any], optional): _description_. Defaults to "interval".
            seconds (int, optional): _description_. Defaults to 5. The amount of seconds to wait before executing the task again (I don't think it is effective for the cron option)
 
        Returns:
            Union[int,Job]: _description_: will raise a ValueError when an error occurs, otherwise, returns an instance of Job.
        """
        if callable(func) is False:
            self.disp.log_error(
                f"The provided function is not callable: {func}.",
                "add_task"
            )
            raise ValueError("The function must be callable.")
        if args is not None and isinstance(args, tuple) is False:
            msg = f"The provided args for {func.__name__} are not tuples.\n"
            msg += f"Converting args: '{args}'  to tuples."
            self.disp.log_warning(msg, "add_task")
            args = tuple((args,))
        if kwargs is not None and isinstance(kwargs, dict) is False:
            msg = f"The provided kwargs for {func.__name__}"
            msg += "are not dictionaries.\n"
            msg += f"Converting kwargs: '{kwargs}' to dictionaries."
            self.disp.log_warning(msg, "add_task")
            kwargs = self._to_dict(kwargs)
            self.disp.log_warning(f"Converted data = {kwargs}.", "add_task")
        if trigger is not None and isinstance(trigger, str) is False:
            self.disp.log_error(
                f"The provided trigger is not a string: {trigger}.",
                "add_task"
            )
            raise ValueError("The trigger must be a string.")
        if isinstance(seconds, int) is False:
            self.disp.log_error(
                f"The provided seconds is not an integer: {seconds}.",
                "add_task"
            )
            raise ValueError("The seconds must be an integer.")
        msg = f"Adding job: {func.__name__} "
        msg += f"with trigger: {trigger}, "
        msg += f"seconds = {seconds}, "
        msg += f"args = {args}, "
        msg += f"kwargs = {kwargs}."
        self.disp.log_debug(msg, "add_task")
        return self.scheduler.add_job(
            func=func,
            trigger=trigger,
            seconds=seconds,
            args=args,
            kwargs=kwargs
        )
 
    def start(self) -> Union[int, None]:
        """_summary_
            The function in charge of starting the scheduler loop.
 
        Raises:
            RuntimeError: _description_: Will raise a runtile error if the underlying functions failled.
 
        Returns:
            Union[int, None]: _description_: Will return self.success if it worked, otherwise None because it will have raised an error.
        """
        try:
            self.scheduler.start()
            self.disp.log_info("Scheduler started...", "start")
            return self.success
        except SchedulerAlreadyRunningError:
            self.disp.log_info("Scheduler is already running...", "start")
            return self.success
        except RuntimeError as e:
            self.disp.log_error(
                f"An error occurred while starting the scheduler: {e}",
                "start"
            )
            msg = f"Error({self.__class__.__name__}): "
            msg += "Failed to call the scheduler's start wrapper function."
            raise RuntimeError(msg) from e
        except Exception as e:
            self.disp.log_error(
                f"An error occurred while starting the scheduler: {e}", "start"
            )
            msg = f"Error({self.__class__.__name__}): "
            msg += "Failed to call the scheduler's start wrapper function."
            raise RuntimeError(msg) from e
 
    def pause(self, pause: bool = True) -> Union[int, None]:
        """_summary_
            This function is in charge of pausing the scheduler if it was running.
 
        Args:
            pause (bool, optional): _description_: This is the boolean that will determine if the scheduler should be paused or not. Defaults to True.
 
        Returns:
            Union[int, None]: _description_: Will return self.success if it worked, otherwise None because it will have raised an error.
        """
        try:
            if pause is True:
                self.scheduler.pause()
                self.disp.log_info("Scheduler paused.", "pause")
            else:
                self.scheduler.resume()
                self.disp.log_info("Scheduler resumed.", "pause")
            return self.success
        except Exception as e:
            self.disp.log_error(
                f"An error occurred while pausing the scheduler: {e}",
                "pause"
            )
            msg = f"Error({self.__class__.__name__}): "
            msg += "Failed to call the chron pause wrapper function."
            raise RuntimeError(msg) from e
 
    def resume(self) -> Union[int, None]:
        """_summary_
            This function is in charge of resuming the scheduler loop if it was paused.
 
        Returns:
            Union[int]: _description_: Will return self.success if it worked, otherwise None because it will have raised an error.
        """
        return self.pause(pause=False)
 
    def stop(self, wait: bool = True) -> Union[int, None]:
        """_summary_
            This function is responsible for shutting down the scheduler, terminating any running jobs, and optionally waiting for those jobs to complete before exiting.
 
        Args:
            wait (bool, optional): _description_. Defaults to True. Wait for the running tasks to finish.
 
        Raises:
            RuntimeError: _description_: The function failed to call the underlying processes that were required for it to run.
 
        Returns:
            Union[int, None]: _description_: will return self.success if it succeeds, or none if it raised an error.
        """
        try:
            self.scheduler.shutdown(wait=wait)
            self.disp.log_info("Scheduler stopped.", "stop")
            return self.success
        except SchedulerNotRunningError:
            self.disp.log_info("Scheduler is already stopped.", "stop")
            return self.success
        except Exception as e:
            self.disp.log_error(
                f"An error occurred while stopping the scheduler: {e}", "stop"
            )
            msg = f"Error({self.__class__.__name__}): "
            msg += "Failed to call the chron stop wrapper function."
            raise RuntimeError(msg) from e
 
 
if __name__ == "__main__":
    import sys
    from time import sleep
    from datetime import datetime
 
    def test_current_date(*args: Any) -> None:
        """_summary_
            This is a test function that will print the current date.
        Args:
            date (datetime): _description_
        """
        if len(args) >= 1:
            date = args[0]
        else:
            date = datetime.now()
        if callable(date):
            print(f"(test_current_date) (Called) Current date: {date()}")
        else:
            print(f"(test_current_date) (Not called) Current date: {date}",)
 
    def hello_world() -> None:
        """_summary_
            This is a test function that will print "Hello, World!"
        """
        print("Hello, World!")
 
    def pending_world() -> None:
        """_summary_
            This is a test function that will print "Pending, World!"
        """
        print("Pending, World!")
 
    def goodbye_world() -> None:
        """_summary_
            This is a test function that will print "Goodbye, World!"
        """
        print("Goodbye, World!")
 
    print("Testing declared functions.")
    test_current_date()
    hello_world()
    pending_world()
    goodbye_world()
    print("Declared functions tested.")
 
    SUCCES = 0
    ERROR = 84
    DEBUG = True
    KIND_KILL = True
    NB_REPEATS = 2
    TRIGGER = "interval"
    SECONDS = 2
    NB_FUNCTIONS = 5
    MAIN_THREAD_DELAY = int((SECONDS*NB_FUNCTIONS)*NB_REPEATS)
 
    print(
        f"Statuses:\nSUCCESS = {SUCCES}, ERROR = {ERROR}\n"
        f"DEBUG = {DEBUG}, KIND_KILL = {KIND_KILL}, "
        f"NB_REPEATS = {NB_REPEATS}, "
        f"TRIGGER = {TRIGGER}, SECONDS = {SECONDS}, "
        f"NB_FUNCTIONS = {NB_FUNCTIONS}, "
        f"MAIN_THREAD_DELAY = {MAIN_THREAD_DELAY}"
    )
 
    print("Initialising class BackgroundTasks.")
    BTI = BackgroundTasks(
        success=SUCCES,
        error=ERROR,
        debug=DEBUG
    )
    print("Class BackgroundTasks initialised.")
 
    print("Adding tasks to the scheduler.")
    status = BTI.safe_add_task(
        func=test_current_date,
        args=(datetime.now,),
        kwargs=None,
        trigger=TRIGGER,
        seconds=SECONDS
    )
    print(f"status {status}")
    status = BTI.add_task(
        hello_world,
        args=None,
        kwargs=None,
        trigger=TRIGGER,
        seconds=SECONDS
    )
    print(f"status {status}")
    status = BTI.safe_add_task(
        pending_world,
        args=None,
        kwargs=None,
        trigger=TRIGGER,
        seconds=SECONDS
    )
    print(f"status {status}")
    status = BTI.add_task(
        goodbye_world,
        args=None,
        kwargs=None,
        trigger=TRIGGER,
        seconds=SECONDS
    )
    print(f"status {status}")
    status = BTI.add_task(
        func=test_current_date,
        args=(datetime.now,),
        kwargs=None,
        trigger=TRIGGER,
        seconds=SECONDS
    )
    print(f"status {status}")
    print("Added tasks to the scheduler.")
 
    print("Startins scheduler.")
    print(f"Status: {BTI.safe_start()}")
    print("Scheduler started.")
    print(f"Waiting {MAIN_THREAD_DELAY} on the main thread.")
    sleep(MAIN_THREAD_DELAY)
    print(f"Waited {MAIN_THREAD_DELAY} on the main thread.")
    print("Stopping scheduler.")
    status = BTI.safe_stop(KIND_KILL)
    print(f"Status: {status}")
    sys.exit(status)