hooks
Hook infrastructure for Bitfount.
Module
Functions
get_hooks
def get_hooks( type: HookType,) ‑> Union[List[bitfount.hooks.AlgorithmHookProtocol], List[bitfount.hooks.PodHookProtocol], List[bitfount.hooks.ProtocolHookProtocol]]:
Get all registered hooks of a particular type.
Arguments
type
: The type of hook to get.
Returns A list of hooks of the provided type.
Raises
ValueError
: If the provided type is not a valid hook type.
Classes
BaseAlgorithmHook
class BaseAlgorithmHook():
Base algorithm hook class.
Initialise the hook.
Ancestors
- bitfount.hooks.BaseHook
Variables
type : [HookType](/api/bitfount/hooks#hooktype)
- Return the hook type.
Methods
on_init_end
def on_init_end(self, algorithm: _BaseAlgorithm, *args: Any, **kwargs: Any) ‑> None:
Run the hook at the very end of algorithm initialisation.
on_init_start
def on_init_start(self, algorithm: _BaseAlgorithm, *args: Any, **kwargs: Any) ‑> None:
Run the hook at the very start of algorithm initialisation.
on_run_end
def on_run_end(self, algorithm: _BaseAlgorithm, *args: Any, **kwargs: Any) ‑> None:
Run the hook at the very end of algorithm run.
on_run_start
def on_run_start(self, algorithm: _BaseAlgorithm, *args: Any, **kwargs: Any) ‑> None:
Run the hook at the very start of algorithm run.
BasePodHook
class BasePodHook():
Base pod hook class.
Initialise the hook.
Ancestors
- bitfount.hooks.BaseHook
Variables
type : [HookType](/api/bitfount/hooks#hooktype)
- Return the hook type.
Methods
on_file_process
def on_file_process( self, datasource: FileSystemIterableSource, file_num: int, total_num_files: int, *args: Any, **kwargs: Any,) ‑> None:
Run the hook when a file is processed.
on_pod_init_end
def on_pod_init_end(self, pod: Pod, *args: Any, **kwargs: Any) ‑> None:
Run the hook at the end of pod initialisation.
on_pod_init_error
def on_pod_init_error( self, pod: Pod, exception: Exception, *args: Any, **kwargs: Any,) ‑> None:
Run the hook if an uncaught exception is raised during pod initialisation.
Raises
NotImplementedError
: If the hook is not implemented. This is to ensure that underlying exceptions are not swallowed if the hook is not implemented. This error is caught further up the chain and the underlying exception is raised instead.
on_pod_init_start
def on_pod_init_start( self, pod: Pod, pod_name: str, username: Optional[str] = None, *args: Any, **kwargs: Any,) ‑> None:
Run the hook at the very start of pod initialisation.
on_pod_shutdown_end
def on_pod_shutdown_end(self, pod: Pod, *args: Any, **kwargs: Any) ‑> None:
Run the hook at the very end of pod shutdown.
on_pod_shutdown_start
def on_pod_shutdown_start(self, pod: Pod, *args: Any, **kwargs: Any) ‑> None:
Run the hook at the very start of pod shutdown.
on_pod_startup_end
def on_pod_startup_end(self, pod: Pod, *args: Any, **kwargs: Any) ‑> None:
Run the hook at the end of pod startup.
on_pod_startup_error
def on_pod_startup_error( self, pod: Pod, exception: Exception, *args: Any, **kwargs: Any,) ‑> None:
Run the hook if an uncaught exception is raised during pod startup.
Raises
NotImplementedError
: If the hook is not implemented. This is to ensure that underlying exceptions are not swallowed if the hook is not implemented. This error is caught further up the chain and the underlying exception is raised instead.
on_pod_startup_start
def on_pod_startup_start(self, pod: Pod, *args: Any, **kwargs: Any) ‑> None:
Run the hook at the very start of pod startup.
on_task_end
def on_task_end(self, pod: Pod, *args: Any, **kwargs: Any) ‑> None:
Run the hook when a new task is received at the end.
on_task_error
def on_task_error( self, pod: Pod, exception: BaseException, *args: Any, **kwargs: Any,) ‑> None:
Run the hook when there is an exception in a task.
on_task_start
def on_task_start(self, pod: Pod, *args: Any, **kwargs: Any) ‑> None:
Run the hook when a new task is received at the start.
HookType
class HookType(value, names=None, *, module=None, qualname=None, type=None, start=1):
Enum for hook types.