Collaboration diagram for src.lib.components.background_tasks.BackgroundTasks:

Public Member Functions
None	__init__ (self, int success=0, int error=84, bool debug=False)

None	__del__ (self)

Union[int, Job]	safe_add_task (self, callable func, Union[Tuple, None] args=None, Union[Dict, None] kwargs=None, Union[str, Any] trigger="interval", int seconds=5)

int	safe_start (self)

int	safe_pause (self, bool pause=True)

int	safe_resume (self)

int	safe_stop (self, bool wait=True)

Union[Job, None]	add_task (self, callable func, Union[Tuple, None] args=None, Union[Dict, None] kwargs=None, Union[str, Any] trigger="interval", int seconds=5)

Union[int, None]	start (self)

Union[int, None]	pause (self, bool pause=True)

Union[int]	resume (self)

Union[int, None]	stop (self, bool wait=True)

Data Fields
int	success = success

int	error = error

bool	debug = debug

	scheduler = BackgroundScheduler()

Disp	disp

Protected Member Functions
dict	_to_dict (self, Union[Any, None] data=None)

Detailed Description

_summary_
    This is the class that is in charge of scheduling background tasks that need to run on intervals

Definition at line 19 of file background_tasks.py.

Constructor & Destructor Documentation

◆ init()

None src.lib.components.background_tasks.BackgroundTasks.__init__	(		self,
		int	success = 0,
		int	error = 84,
		bool	debug = False )

Definition at line 24 of file background_tasks.py.

◆ del()

None src.lib.components.background_tasks.BackgroundTasks.__del__ ( self )

_summary_
    The destructor of the class

Definition at line 41 of file background_tasks.py.

Member Function Documentation

◆ _to_dict()

dict src.lib.components.background_tasks.BackgroundTasks._to_dict	(		self,
		Union[Any, None]	data = None )

protected

_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.

Definition at line 152 of file background_tasks.py.

◆ add_task()

Union[Job, None] src.lib.components.background_tasks.BackgroundTasks.add_task	(		self,
		callable	func,
		Union[Tuple, None]	args = None,
		Union[Dict, None]	kwargs = None,
		Union[str, Any]	trigger = "interval",
		int	seconds = 5 )

_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.

Definition at line 172 of file background_tasks.py.

◆ pause()

Union[int, None] src.lib.components.background_tasks.BackgroundTasks.pause	(		self,
		bool	pause = True )

_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.

Definition at line 264 of file background_tasks.py.

◆ resume()

Union[int] src.lib.components.background_tasks.BackgroundTasks.resume ( self )

_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.

Definition at line 291 of file background_tasks.py.

◆ safe_add_task()

Union[int, Job] src.lib.components.background_tasks.BackgroundTasks.safe_add_task	(		self,
		callable	func,
		Union[Tuple, None]	args = None,
		Union[Dict, None]	kwargs = None,
		Union[str, Any]	trigger = "interval",
		int	seconds = 5 )

_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.

Definition at line 53 of file background_tasks.py.

◆ safe_pause()

int src.lib.components.background_tasks.BackgroundTasks.safe_pause	(		self,
		bool	pause = True )

_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

Definition at line 98 of file background_tasks.py.

◆ safe_resume()

int src.lib.components.background_tasks.BackgroundTasks.safe_resume ( self )

_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.

Definition at line 117 of file background_tasks.py.

◆ safe_start()

int src.lib.components.background_tasks.BackgroundTasks.safe_start ( self )

_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.

Definition at line 82 of file background_tasks.py.

◆ safe_stop()

int src.lib.components.background_tasks.BackgroundTasks.safe_stop	(		self,
		bool	wait = True )

_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

Definition at line 133 of file background_tasks.py.

◆ start()

Union[int, None] src.lib.components.background_tasks.BackgroundTasks.start ( self )

_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.

Definition at line 231 of file background_tasks.py.

◆ stop()

Union[int, None] src.lib.components.background_tasks.BackgroundTasks.stop	(		self,
		bool	wait = True )

_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.

Definition at line 300 of file background_tasks.py.

Field Documentation

◆ debug

bool src.lib.components.background_tasks.BackgroundTasks.debug = debug

Definition at line 28 of file background_tasks.py.

◆ disp

Disp src.lib.components.background_tasks.BackgroundTasks.disp

Initial value:

=  Disp(
            TOML_CONF,
            FILE_DESCRIPTOR,
            SAVE_TO_FILE,
            FILE_NAME,
            debug=self.debug,
            logger=self.__class__.__name__
        )

Definition at line 32 of file background_tasks.py.

◆ error

int src.lib.components.background_tasks.BackgroundTasks.error = error

Definition at line 27 of file background_tasks.py.

◆ scheduler

src.lib.components.background_tasks.BackgroundTasks.scheduler = BackgroundScheduler()

Definition at line 30 of file background_tasks.py.

◆ success

int src.lib.components.background_tasks.BackgroundTasks.success = success

Definition at line 26 of file background_tasks.py.

The documentation for this class was generated from the following file:

app/back/server/src/lib/components/background_tasks.py

Public Member Functions

Data Fields

Protected Member Functions