API Reference

This page is automatically generated from the source code docstrings.

Manages and executes a collection of interdependent tasks.

A Process orchestrates the execution of multiple tasks, handling dependency resolution, task ordering. Task execution can be performed in parallel or sequentially. It provides logging management and error propagation for dependent tasks. If a task fails, all tasks depending on it are marked as failed without execution, but non-dependent tasks continue to run.

Attributes

tasks : list[Task] List of tasks to be executed, automatically sorted by dependencies. runner : ProcessRunner The runner responsible for executing the tasks.

Raises

TypeError If tasks is not a list or contains non-Task elements. ValueError If duplicate task names are found. DependencyNotFoundError If a task depends on a non-existent task. CircularDependencyError If circular dependencies are detected among tasks.

Source code in src/processes/process.py

class Process:
    """
    Manages and executes a collection of interdependent tasks.

    A Process orchestrates the execution of multiple tasks, handling dependency
    resolution, task ordering. Task execution can be performed in parallel or sequentially. It
    provides logging management and error propagation for dependent tasks. If a task fails,
    all tasks depending on it are marked as failed without execution, but non-dependent tasks
    continue to run.

    Attributes
    ----------
    tasks : list[Task]
        List of tasks to be executed, automatically sorted by dependencies.
    runner : ProcessRunner
        The runner responsible for executing the tasks.

    Raises
    ------
    TypeError
        If tasks is not a list or contains non-Task elements.
    ValueError
        If duplicate task names are found.
    DependencyNotFoundError
        If a task depends on a non-existent task.
    CircularDependencyError
        If circular dependencies are detected among tasks.
    """

    def __init__(self, tasks: list[Task]):
        self.tasks = tasks

        try:
            self._check_input_types()
            self._check_duplicate_names()
            self._check_dependencies_exist()
            self._topological_sort()
        except Exception as e:
            self.close_loggers()
            raise e
        self.runner = ProcessRunner(self)

    def __enter__(self) -> Self:
        """Called when entering the 'with' block."""
        return self

    def __exit__(
        self,
        exc_type: type[BaseException] | None,
        exc_value: BaseException | None,
        traceback: TracebackType | None,
    ) -> Literal[False]:
        """Called when exiting the 'with' block, even if an error occurred."""
        self.close_loggers()
        return False

    def _check_input_types(self) -> None:
        """Validate that tasks is a list containing only Task objects.

        Raises
        ------
        TypeError
            If tasks is not a list or contains non-Task elements.
        """
        if not isinstance(self.tasks, list):
            raise TypeError(f"tasks must be list. Got {type(self.tasks)}")
        for task in self.tasks:
            if not isinstance(task, Task):
                raise TypeError(f"task must be Task. Got {type(task)}")

    def _check_duplicate_names(self) -> None:
        """Verify that all task names are unique.

        Raises
        ------
        ValueError
            If duplicate task names are found.
        """
        names = set()
        for task in self.tasks:
            if task.name in names:
                raise ValueError(f"Duplicate task name: {task.name}")
            names.add(task.name)

    def _check_dependencies_exist(self) -> None:
        """Verify that all task dependencies refer to existing tasks.

        Raises
        ------
        DependencyNotFoundError
            If a task depends on a non-existent task.
        """
        names = {t.name for t in self.tasks}
        for task in self.tasks:
            for dep in task.get_dependencies_names():
                if dep not in names:
                    raise DependencyNotFoundError(
                        f"Task {task.name} depends on missing task: {dep}"
                    )

    def _topological_sort(self) -> None:
        """Sort tasks based on dependencies using Kahn's Algorithm in O(V+E) time.

        Reorders the task list so that dependencies are always executed before
        tasks that depend on them.

        Raises
        ------
        CircularDependencyError
            If circular dependencies are detected among tasks.
        """
        in_degree = {t.name: 0 for t in self.tasks}
        graph: dict[str, list[str]] = {t.name: [] for t in self.tasks}
        task_map = {t.name: t for t in self.tasks}

        for task in self.tasks:
            for dep in task.dependencies:
                graph[dep.task_name].append(task.name)
                in_degree[task.name] += 1

        queue = [name for name, deg in in_degree.items() if deg == 0]
        sorted_tasks = []

        while queue:
            u = queue.pop(0)
            sorted_tasks.append(task_map[u])
            for v in graph[u]:
                in_degree[v] -= 1
                if in_degree[v] == 0:
                    queue.append(v)

        if len(sorted_tasks) != len(self.tasks):
            raise CircularDependencyError("Circular dependency detected.")
        self.tasks = sorted_tasks

    def get_task(self, task_name: str) -> Task:
        """Retrieve a task by name.

        Parameters
        ----------
        task_name : str
            The name of the task to retrieve.

        Returns
        -------
        Task
            The task with the specified name.

        Raises
        ------
        TaskNotFoundError
            If no task with the given name exists.
        """
        for task in self.tasks:
            if task.name == task_name:
                return task
        raise TaskNotFoundError(f"Task not found: {task_name}")

    def run(self, parallel: bool | None = None, max_workers: int = 4) -> ProcessResult:
        """Execute all tasks in the process.

        Runs tasks sequentially or in parallel while respecting dependencies.
        Dependencies are always resolved before dependent tasks are executed.

        Parameters
        ----------
        parallel : bool, optional
            Whether to run tasks in parallel while respecting dependencies.
            If None, automatically set to True for processes with 10 or more tasks,
            False otherwise. Defaults to None.
        max_workers : int, optional
            Maximum number of worker threads for parallel execution. Defaults to 4.
            Only used when parallel=True. If set to 1, falls back to sequential execution.

        Returns
        -------
        ProcessResult
            Contains passed_tasks_results (dict mapping task names to TaskResult)
            and failed_tasks (set of task names that failed).
        """
        if parallel is None:
            parallel = len(self.tasks) >= 10

        max_workers = max(1, max_workers)
        if parallel:
            if max_workers == 1:
                parallel = False  # Fallback to sequential if only one worker
        process_result = self.runner.run(parallel, max_workers)
        return process_result

    def get_dependant_tasks(self, task_name: str) -> list[Task]:
        """Retrieve all tasks that directly or indirectly depend on a given task.

        Parameters
        ----------
        task_name : str
            The name of the task to find dependants for.

        Returns
        -------
        list[Task]
            List of all tasks that depend on the specified task, including
            transitive dependencies (tasks that depend on tasks that depend
            on the specified task).
        """
        found = []

        def find(name: str) -> None:
            for t in self.tasks:
                if name in t.get_dependencies_names() and t not in found:
                    found.append(t)
                    find(t.name)

        find(task_name)
        return found

    def close_loggers(self) -> None:
        """Close and clean up all logger handlers for all tasks.

        Should be called when the process is done to ensure proper resource cleanup.
        """
        for task in self.tasks:
            for handler in task.logger.handlers:
                handler.close()
                task.logger.removeHandler(handler)

__enter__()

Called when entering the 'with' block.

Source code in src/processes/process.py

def __enter__(self) -> Self:
    """Called when entering the 'with' block."""
    return self

__exit__(exc_type, exc_value, traceback)

Called when exiting the 'with' block, even if an error occurred.

Source code in src/processes/process.py

def __exit__(
    self,
    exc_type: type[BaseException] | None,
    exc_value: BaseException | None,
    traceback: TracebackType | None,
) -> Literal[False]:
    """Called when exiting the 'with' block, even if an error occurred."""
    self.close_loggers()
    return False

close_loggers()

Close and clean up all logger handlers for all tasks.

Should be called when the process is done to ensure proper resource cleanup.

Source code in src/processes/process.py

def close_loggers(self) -> None:
    """Close and clean up all logger handlers for all tasks.

    Should be called when the process is done to ensure proper resource cleanup.
    """
    for task in self.tasks:
        for handler in task.logger.handlers:
            handler.close()
            task.logger.removeHandler(handler)

get_dependant_tasks(task_name)

Retrieve all tasks that directly or indirectly depend on a given task.

Parameters

task_name : str The name of the task to find dependants for.

Returns

list[Task] List of all tasks that depend on the specified task, including transitive dependencies (tasks that depend on tasks that depend on the specified task).

Source code in src/processes/process.py

def get_dependant_tasks(self, task_name: str) -> list[Task]:
    """Retrieve all tasks that directly or indirectly depend on a given task.

    Parameters
    ----------
    task_name : str
        The name of the task to find dependants for.

    Returns
    -------
    list[Task]
        List of all tasks that depend on the specified task, including
        transitive dependencies (tasks that depend on tasks that depend
        on the specified task).
    """
    found = []

    def find(name: str) -> None:
        for t in self.tasks:
            if name in t.get_dependencies_names() and t not in found:
                found.append(t)
                find(t.name)

    find(task_name)
    return found

get_task(task_name)

Retrieve a task by name.

Parameters

task_name : str The name of the task to retrieve.

Returns

Task The task with the specified name.

Raises

TaskNotFoundError If no task with the given name exists.

Source code in src/processes/process.py

def get_task(self, task_name: str) -> Task:
    """Retrieve a task by name.

    Parameters
    ----------
    task_name : str
        The name of the task to retrieve.

    Returns
    -------
    Task
        The task with the specified name.

    Raises
    ------
    TaskNotFoundError
        If no task with the given name exists.
    """
    for task in self.tasks:
        if task.name == task_name:
            return task
    raise TaskNotFoundError(f"Task not found: {task_name}")

run(parallel=None, max_workers=4)

Execute all tasks in the process.

Runs tasks sequentially or in parallel while respecting dependencies. Dependencies are always resolved before dependent tasks are executed.

Parameters

parallel : bool, optional Whether to run tasks in parallel while respecting dependencies. If None, automatically set to True for processes with 10 or more tasks, False otherwise. Defaults to None. max_workers : int, optional Maximum number of worker threads for parallel execution. Defaults to 4. Only used when parallel=True. If set to 1, falls back to sequential execution.

Returns

ProcessResult Contains passed_tasks_results (dict mapping task names to TaskResult) and failed_tasks (set of task names that failed).

Source code in src/processes/process.py

def run(self, parallel: bool | None = None, max_workers: int = 4) -> ProcessResult:
    """Execute all tasks in the process.

    Runs tasks sequentially or in parallel while respecting dependencies.
    Dependencies are always resolved before dependent tasks are executed.

    Parameters
    ----------
    parallel : bool, optional
        Whether to run tasks in parallel while respecting dependencies.
        If None, automatically set to True for processes with 10 or more tasks,
        False otherwise. Defaults to None.
    max_workers : int, optional
        Maximum number of worker threads for parallel execution. Defaults to 4.
        Only used when parallel=True. If set to 1, falls back to sequential execution.

    Returns
    -------
    ProcessResult
        Contains passed_tasks_results (dict mapping task names to TaskResult)
        and failed_tasks (set of task names that failed).
    """
    if parallel is None:
        parallel = len(self.tasks) >= 10

    max_workers = max(1, max_workers)
    if parallel:
        if max_workers == 1:
            parallel = False  # Fallback to sequential if only one worker
    process_result = self.runner.run(parallel, max_workers)
    return process_result

Container for the results of a process execution.

Holds the outcomes of all tasks executed in a process, separating successful and failed tasks with their respective results.

Attributes

passed_tasks_results : dict[str, TaskResult] Mapping of task names to TaskResult objects for all tasks that executed successfully. failed_tasks : set[str] Set of task names for all tasks that failed during execution.

Source code in src/processes/process.py

class ProcessResult:
    """
    Container for the results of a process execution.

    Holds the outcomes of all tasks executed in a process, separating successful
    and failed tasks with their respective results.

    Attributes
    ----------
    passed_tasks_results : dict[str, TaskResult]
        Mapping of task names to TaskResult objects for all tasks that executed successfully.
    failed_tasks : set[str]
        Set of task names for all tasks that failed during execution.
    """

    def __init__(self, passed_tasks_results: dict[str, TaskResult], failed_tasks: set[str]):
        self.passed_tasks_results = passed_tasks_results
        self.failed_tasks = failed_tasks

A Task represents a unit of work to be executed within a Process.

A Task encapsulates a callable function with its arguments, dependencies on other tasks, and logging configuration. Tasks can be executed, by the Process class, sequentially or in parallel, with automatic dependency resolution and result passing between dependent tasks.

Attributes

name : str Unique name for the task (cannot contain spaces). log_path : str File path where task logs will be written. func : Callable The function to execute when the task runs. args : tuple Positional arguments to pass to the function. Defaults to empty tuple. kwargs : dict Keyword arguments to pass to the function. Defaults to empty dict. dependencies : list[TaskDependency] List of tasks this task depends on. Defaults to empty list. html_mail_handler : HTMLSMTPHandler, optional Handler for sending error logs via email in HTML format. Defaults to None. logger : logging.Logger Logger instance for this task, automatically configured.

Source code in src/processes/task.py

class Task:
    """
    A Task represents a unit of work to be executed within a Process.

    A Task encapsulates a callable function with its arguments, dependencies on other tasks,
    and logging configuration. Tasks can be executed, by the Process class, sequentially
    or in parallel, with automatic dependency resolution and result passing between dependent tasks.

    Attributes
    ----------
    name : str
        Unique name for the task (cannot contain spaces).
    log_path : str
        File path where task logs will be written.
    func : Callable
        The function to execute when the task runs.
    args : tuple
        Positional arguments to pass to the function. Defaults to empty tuple.
    kwargs : dict
        Keyword arguments to pass to the function. Defaults to empty dict.
    dependencies : list[TaskDependency]
        List of tasks this task depends on. Defaults to empty list.
    html_mail_handler : HTMLSMTPHandler, optional
        Handler for sending error logs via email in HTML format. Defaults to None.
    logger : logging.Logger
        Logger instance for this task, automatically configured.
    """

    kwargs: dict[str, Any]
    dependencies: list[TaskDependency]

    def __init__(
        self,
        name: str,
        log_path: str,
        func: Callable[..., Any],
        args: tuple[Any, ...] = (),
        kwargs: dict[str, Any] | None = None,
        dependencies: list[TaskDependency] | None = None,
        html_mail_handler: HTMLSMTPHandler | None = None,
    ):
        self.name = name
        self.log_path = log_path
        self.func = func
        self.args = args
        self.html_mail_handler = html_mail_handler

        if kwargs is None:
            self.kwargs = {}
        else:
            self.kwargs = kwargs
        if dependencies is None:
            self.dependencies = []
        else:
            self.dependencies = dependencies

        self._check_input_types()
        if " " in self.name:
            raise ValueError(f"Task name cannot contain spaces. Got {self.name}")

        depedencies_names = []
        for dependency in self.dependencies:
            if dependency.task_name in depedencies_names:
                raise ValueError(f"Duplicate dependency name: {dependency.task_name}")
            depedencies_names.append(dependency.task_name)
            if dependency.task_name == self.name:
                raise ValueError(
                    f"Got dependency with same name as Task. "
                    f"Task: {self.name}. Dependency: {dependency.task_name}"
                )

        logger = logging.getLogger(self.name)
        logger.setLevel(logging.DEBUG)
        if logger.hasHandlers():
            logger.handlers.clear()

        file_handler = logging.FileHandler(self.log_path)
        file_handler.setLevel(logging.INFO)
        formatter = logging.Formatter("%(asctime)s - %(name)s - %(levelname)s - %(message)s")
        file_handler.setFormatter(formatter)
        logger.addHandler(file_handler)

        if self.html_mail_handler is not None:
            _html_mail_handler = self.html_mail_handler.copy()
            _html_mail_handler.setFormatter(ExceptionHTMLFormatter())
            _html_mail_handler.setLevel(logging.ERROR)
            _html_mail_handler.subject = f"Error in task {self.name}"
            logger.addHandler(_html_mail_handler)

        self.logger = logger

    def _check_input_types(self) -> None:
        """
        Validates all input parameter types.

        Raises
        ------
        TypeError
            If any parameter is not of the expected type.
        """
        if not callable(self.func):
            raise TypeError(f"func must be callable. Got {type(self.func)}")

        if not isinstance(self.args, tuple):
            raise TypeError(f"args must be tuple. Got {type(self.args)}")

        if not isinstance(self.kwargs, dict):
            raise TypeError(f"kwargs must be dict. Got {type(self.kwargs)}")

        if self.html_mail_handler is not None and not isinstance(
            self.html_mail_handler, HTMLSMTPHandler
        ):
            raise TypeError(
                f"mail_cfg must be of type HTMLSMTPHandler. Got {type(self.html_mail_handler)}"
            )

        if not isinstance(self.dependencies, list):
            raise TypeError(f"dependencies must be list. Got {type(self.dependencies)}")

        for dependency in self.dependencies:
            if not isinstance(dependency, TaskDependency):
                raise TypeError(
                    f"dependency must be of type TaskDependency. Got {type(dependency)}"
                )

    def get_dependencies_names(self) -> set[str]:
        """
        Get the names of all tasks this task depends on.

        Returns
        -------
        set[str]
            Set of dependency task names.
        """
        return {dependency.task_name for dependency in self.dependencies}

    def run(self, executing_process: Process | None = None) -> TaskResult:
        """
        Execute the task's function with its arguments and dependencies.

        This method runs the task's function, automatically injecting results from
        dependent tasks as specified in the dependency configuration. Logs the task
        execution and captures any exceptions.

        Parameters
        ----------
        executing_process : Process, optional
            The parent Process executing this task. Used to retrieve results from
            dependent tasks. Defaults to None.

        Returns
        -------
        TaskResult
            Object containing:
            - worked (bool): True if execution succeeded, False otherwise.
            - result: The return value of the function if successful, None if failed.
            - exception (Exception | None): The exception raised if execution failed,
            None if successful.
        """
        final_args = list(self.args)  # Start with original positional args
        final_kwargs = self.kwargs.copy()  # Start with original keyword args

        if executing_process is not None:
            for dep in self.dependencies:
                dep_result = executing_process.runner.passed_results[dep.task_name].result
                if dep.use_result_as_additional_args:
                    final_args.append(dep_result)
                if dep.use_result_as_additional_kwargs:
                    final_kwargs[dep.additional_kwarg_name] = dep_result

        try:
            self.logger.info(f"Starting {self.name}.")
            result = self.func(*final_args, **final_kwargs)
            self.logger.info(f"Finished {self.name}.")
            return TaskResult(True, result, None)
        except Exception as e:
            report = ""
            if executing_process is not None:
                dependencies_names = [
                    d.name for d in executing_process.get_dependant_tasks(self.name)
                ]
                if dependencies_names:
                    report = (
                        "<h3>Downstream Impact</h3><p>The following tasks will be skipped:</p><ul>"
                    )
                    report += "".join(
                        f"<li>{dependency_name}</li>" for dependency_name in dependencies_names
                    )
                    report += "</ul>"
            report += f"<p><b>Context:</b><br>Function: {self.func.__name__}"
            report += f"<br>Args: {self.args}<br>Kwargs: {self.kwargs}</p>"
            self.logger.exception(e, extra={"post_traceback_html_body": report})
            return TaskResult(False, None, e)

get_dependencies_names()

Get the names of all tasks this task depends on.

Returns

set[str] Set of dependency task names.

Source code in src/processes/task.py

def get_dependencies_names(self) -> set[str]:
    """
    Get the names of all tasks this task depends on.

    Returns
    -------
    set[str]
        Set of dependency task names.
    """
    return {dependency.task_name for dependency in self.dependencies}

run(executing_process=None)

Execute the task's function with its arguments and dependencies.

This method runs the task's function, automatically injecting results from dependent tasks as specified in the dependency configuration. Logs the task execution and captures any exceptions.

Parameters

executing_process : Process, optional The parent Process executing this task. Used to retrieve results from dependent tasks. Defaults to None.

Returns

TaskResult Object containing: - worked (bool): True if execution succeeded, False otherwise. - result: The return value of the function if successful, None if failed. - exception (Exception | None): The exception raised if execution failed, None if successful.

Source code in src/processes/task.py

def run(self, executing_process: Process | None = None) -> TaskResult:
    """
    Execute the task's function with its arguments and dependencies.

    This method runs the task's function, automatically injecting results from
    dependent tasks as specified in the dependency configuration. Logs the task
    execution and captures any exceptions.

    Parameters
    ----------
    executing_process : Process, optional
        The parent Process executing this task. Used to retrieve results from
        dependent tasks. Defaults to None.

    Returns
    -------
    TaskResult
        Object containing:
        - worked (bool): True if execution succeeded, False otherwise.
        - result: The return value of the function if successful, None if failed.
        - exception (Exception | None): The exception raised if execution failed,
        None if successful.
    """
    final_args = list(self.args)  # Start with original positional args
    final_kwargs = self.kwargs.copy()  # Start with original keyword args

    if executing_process is not None:
        for dep in self.dependencies:
            dep_result = executing_process.runner.passed_results[dep.task_name].result
            if dep.use_result_as_additional_args:
                final_args.append(dep_result)
            if dep.use_result_as_additional_kwargs:
                final_kwargs[dep.additional_kwarg_name] = dep_result

    try:
        self.logger.info(f"Starting {self.name}.")
        result = self.func(*final_args, **final_kwargs)
        self.logger.info(f"Finished {self.name}.")
        return TaskResult(True, result, None)
    except Exception as e:
        report = ""
        if executing_process is not None:
            dependencies_names = [
                d.name for d in executing_process.get_dependant_tasks(self.name)
            ]
            if dependencies_names:
                report = (
                    "<h3>Downstream Impact</h3><p>The following tasks will be skipped:</p><ul>"
                )
                report += "".join(
                    f"<li>{dependency_name}</li>" for dependency_name in dependencies_names
                )
                report += "</ul>"
        report += f"<p><b>Context:</b><br>Function: {self.func.__name__}"
        report += f"<br>Args: {self.args}<br>Kwargs: {self.kwargs}</p>"
        self.logger.exception(e, extra={"post_traceback_html_body": report})
        return TaskResult(False, None, e)

Represents a dependency relationship between tasks.

Defines how a task depends on another task, including how the result of the dependency should be passed to the dependent task (as additional positional arguments, keyword arguments, or both).

Attributes

task_name : str The name of the task this dependency refers to. use_result_as_additional_args : bool If True, the result of the dependency task will be passed as an additional positional argument as the last argument. Defaults to False. use_result_as_additional_kwargs : bool If True, the result of the dependency task will be passed as a keyword argument. Defaults to False. additional_kwarg_name : str | None The name of the keyword argument to use if use_result_as_additional_kwargs is True. Required when use_result_as_additional_kwargs is True. Defaults to None.

Raises

TypeError If any parameter type is invalid or if use_result_as_additional_kwargs is True but additional_kwarg_name is not a string.

Source code in src/processes/task.py

class TaskDependency:
    """
    Represents a dependency relationship between tasks.

    Defines how a task depends on another task, including how the result
    of the dependency should be passed to the dependent task (as additional
    positional arguments, keyword arguments, or both).

    Attributes
    ----------
    task_name : str
        The name of the task this dependency refers to.
    use_result_as_additional_args : bool
        If True, the result of the dependency task will be passed as an
        additional positional argument as the last argument. Defaults to False.
    use_result_as_additional_kwargs : bool
        If True, the result of the dependency task will be passed as a
        keyword argument. Defaults to False.
    additional_kwarg_name : str | None
        The name of the keyword argument to use if use_result_as_additional_kwargs
        is True. Required when use_result_as_additional_kwargs is True.
        Defaults to None.

    Raises
    ------
    TypeError
        If any parameter type is invalid or if use_result_as_additional_kwargs
        is True but additional_kwarg_name is not a string.
    """

    def __init__(
        self,
        task_name: str,
        use_result_as_additional_args: bool = False,
        use_result_as_additional_kwargs: bool = False,
        additional_kwarg_name: str = "",
    ):
        self.task_name = task_name
        self.use_result_as_additional_args = use_result_as_additional_args
        self.use_result_as_additional_kwargs = use_result_as_additional_kwargs
        self.additional_kwarg_name = additional_kwarg_name

        if not isinstance(self.task_name, str):
            raise TypeError(f"task_name must be of type str. Got {type(self.task_name)}")
        if not isinstance(self.use_result_as_additional_args, bool):
            raise TypeError(
                f"use_result_as_additional_args must be of type bool. "
                f"Got {type(self.use_result_as_additional_args)}"
            )
        if not isinstance(self.use_result_as_additional_kwargs, bool):
            raise TypeError(
                f"use_result_as_additional_kwargs must be of type bool. "
                f"Got {type(self.use_result_as_additional_kwargs)}"
            )

        if self.use_result_as_additional_kwargs and self.additional_kwarg_name == "":
            raise TypeError(
                "If use_result_as_additional_kwargs is True, additional_kwarg_name"
                " must be a non-empty string."
            )

    def __hash__(self) -> int:
        """
        Return hash of the dependency based on task name.

        Returns
        -------
        int
            Hash value based on the task_name attribute.
        """
        return hash(self.task_name)

__hash__()

Return hash of the dependency based on task name.

Returns

int Hash value based on the task_name attribute.

Source code in src/processes/task.py

def __hash__(self) -> int:
    """
    Return hash of the dependency based on task name.

    Returns
    -------
    int
        Hash value based on the task_name attribute.
    """
    return hash(self.task_name)

Bases: SMTPHandler

A logging handler that sends log records via SMTP as HTML formatted emails.

Extends the standard SMTPHandler to support HTML-formatted email messages, enabling richer formatting and styling in error notifications.

Attributes

mailhost : tuple[str, int] A tuple of (host, port) for the SMTP server. fromaddr : str The email address to send messages from. toaddrs : list[str] List of email addresses to send messages to. credentials : tuple[str, str] | None A tuple of (username, password) for SMTP authentication. Defaults to None. secure : tuple | tuple[str, str] | tuple[str, str, ssl.SSLContext] | None Security configuration for SMTP connection. Can be an empty tuple for no security, a tuple of (certfile, keyfile), or (certfile, keyfile, SSLContext). Defaults to None. timeout : int Connection timeout in seconds. Defaults to 5.

Source code in src/processes/html_logging.py

class HTMLSMTPHandler(logging.handlers.SMTPHandler):
    """
    A logging handler that sends log records via SMTP as HTML formatted emails.

    Extends the standard SMTPHandler to support HTML-formatted email messages,
    enabling richer formatting and styling in error notifications.

    Attributes
    ----------
    mailhost : tuple[str, int]
        A tuple of (host, port) for the SMTP server.
    fromaddr : str
        The email address to send messages from.
    toaddrs : list[str]
        List of email addresses to send messages to.
    credentials : tuple[str, str] | None
        A tuple of (username, password) for SMTP authentication. Defaults to None.
    secure : tuple | tuple[str, str] | tuple[str, str, ssl.SSLContext] | None
        Security configuration for SMTP connection. Can be an empty tuple for no security,
        a tuple of (certfile, keyfile), or (certfile, keyfile, SSLContext).
        Defaults to None.
    timeout : int
        Connection timeout in seconds. Defaults to 5.
    """

    def __init__(
        self,
        mailhost: tuple[str, int],
        fromaddr: str,
        toaddrs: list[str],
        credentials: tuple[str, str] | None = None,
        secure: tuple[()]
        | tuple[str]
        | tuple[str, str]
        | tuple[str, str, ssl.SSLContext]
        | None = None,
        timeout: int = 5,
    ):
        self._crd = credentials
        self._sec = secure
        self._to = timeout

        super().__init__(
            mailhost,
            fromaddr,
            toaddrs,
            "",
            credentials=credentials,
            secure=secure,  # type: ignore[arg-type]
            timeout=timeout,
        )

    def copy(self) -> "HTMLSMTPHandler":
        """Create a shallow copy of this handler.

        Returns
        -------
        HTMLSMTPHandler
            A new HTMLSMTPHandler instance with the same configuration.
        """
        return HTMLSMTPHandler(
            self.mailhost,  # type: ignore[arg-type]
            self.fromaddr,
            self.toaddrs,
            credentials=self._crd,
            secure=self._sec,
            timeout=self._to,
        )

    def __copy__(self) -> "HTMLSMTPHandler":
        """Support for copy.copy() method.

        Returns
        -------
        HTMLSMTPHandler
            A shallow copy of this handler.
        """
        return self.copy()

    def emit(self, record: logging.LogRecord) -> None:
        """Send a log record via email as HTML formatted message.

        Formats the log record using the handler's formatter and sends it
        as an HTML-formatted email. Errors during sending are handled gracefully.

        Parameters
        ----------
        record : logging.LogRecord
            The log record to send.
        """
        try:
            port = self.mailport
            if not port:
                port = smtplib.SMTP_PORT
            smtp = smtplib.SMTP(self.mailhost, port)
            msg = self.format(record)

            # Create MIMEText object with HTML content
            mime_msg = MIMEText(msg, "html")
            mime_msg["From"] = self.fromaddr
            mime_msg["To"] = ",".join(self.toaddrs)
            mime_msg["Subject"] = self.getSubject(record)
            mime_msg["Date"] = formatdate()

            if self.username:
                if self.secure is not None:
                    smtp.starttls(*self.secure)
                smtp.login(self.username, self.password)
            smtp.sendmail(self.fromaddr, self.toaddrs, mime_msg.as_string())
            smtp.quit()
        except Exception:
            self.handleError(record)

__copy__()

Support for copy.copy() method.

Returns

HTMLSMTPHandler A shallow copy of this handler.

Source code in src/processes/html_logging.py

def __copy__(self) -> "HTMLSMTPHandler":
    """Support for copy.copy() method.

    Returns
    -------
    HTMLSMTPHandler
        A shallow copy of this handler.
    """
    return self.copy()

copy()

Create a shallow copy of this handler.

Returns

HTMLSMTPHandler A new HTMLSMTPHandler instance with the same configuration.

Source code in src/processes/html_logging.py

def copy(self) -> "HTMLSMTPHandler":
    """Create a shallow copy of this handler.

    Returns
    -------
    HTMLSMTPHandler
        A new HTMLSMTPHandler instance with the same configuration.
    """
    return HTMLSMTPHandler(
        self.mailhost,  # type: ignore[arg-type]
        self.fromaddr,
        self.toaddrs,
        credentials=self._crd,
        secure=self._sec,
        timeout=self._to,
    )

emit(record)

Send a log record via email as HTML formatted message.

Formats the log record using the handler's formatter and sends it as an HTML-formatted email. Errors during sending are handled gracefully.

Parameters

record : logging.LogRecord The log record to send.

Source code in src/processes/html_logging.py

def emit(self, record: logging.LogRecord) -> None:
    """Send a log record via email as HTML formatted message.

    Formats the log record using the handler's formatter and sends it
    as an HTML-formatted email. Errors during sending are handled gracefully.

    Parameters
    ----------
    record : logging.LogRecord
        The log record to send.
    """
    try:
        port = self.mailport
        if not port:
            port = smtplib.SMTP_PORT
        smtp = smtplib.SMTP(self.mailhost, port)
        msg = self.format(record)

        # Create MIMEText object with HTML content
        mime_msg = MIMEText(msg, "html")
        mime_msg["From"] = self.fromaddr
        mime_msg["To"] = ",".join(self.toaddrs)
        mime_msg["Subject"] = self.getSubject(record)
        mime_msg["Date"] = formatdate()

        if self.username:
            if self.secure is not None:
                smtp.starttls(*self.secure)
            smtp.login(self.username, self.password)
        smtp.sendmail(self.fromaddr, self.toaddrs, mime_msg.as_string())
        smtp.quit()
    except Exception:
        self.handleError(record)