Handle timeout using multiprocessing

timeout_handler.py

import multiprocessing
import time
from typing import Callable, Union


class TimeoutHandler:
    """Initiates TimeoutHandler object to run a function and terminate it after a given time limit.

    >>> TimeoutHandler

    """

    def __init__(self, timeout: Union[int, float]):
        """Instantiates necessary args.

        Args:
            timeout: Timeout after which the said function has to be terminated.
        """
        self.timeout = timeout

    def run(self, function: Callable, *args, **kwargs) -> str:
        """Runs the given function.

        Args:
            function: Function to run and timeout.
            *args: Args to be passed to the function.
            **kwargs: Keyword args to be passed to the function.

        Returns:
            str:
            Status if the function has completed within the given timeout or not.
        """
        proc = multiprocessing.Process(target=function, args=args, kwargs=kwargs)
        _start = time.time()
        proc.start()
        proc.join(timeout=self.timeout)
        if proc.is_alive():
            proc.kill()
            return f"Function [{function.__name__}] exceeded {self.timeout} seconds. Process PID [{proc.pid}] killed!"
        return f"Function [{function.__name__}] finished in {round(float(time.time() - _start), 2)} seconds"


def sleeper(interval, name):
    time.sleep(interval)
    print("Successful")


if __name__ == '__main__':
    start = time.time()
    print(TimeoutHandler(timeout=5).run(function=sleeper, **{"interval": 3, "name": "Vicky"}))
    print(time.time() - start)