4
\$\begingroup\$

I've written a library with convenience functions for working with the Python threading module (specifically threading.Thread and threading.Timer). I'd appreciate it if someone could take a look at it. I've also uploaded the code on GitHub.


"""
This module contains various decorators to help with common tasks involving
the Python `threading` module.

- `thread`: Call a function in a separate thread.

- `thread_join`: Call a function in a separate thread and join the thread.

- `thread_daemon`: Call a function in a separate thread and return
its return value.

- `delay`: Delay before calling a function.

- `delay_join`: Delay before calling a function and join the thread.

- `delay_daemon`: Delay before calling a function in a daemon thread.

- `delay_with_return_value`: Delay before calling a function and return
its return value.
"""

import threading

__all__ = [
    "thread",
    "thread_join",
    "thread_daemon",
    "thread_with_return_value",
    "delay",
    "delay_join",
    "delay_daemon",
    "delay_with_return_value"
]


class _FunctionThreadWithReturnValue(threading.Thread):
    """Calls a function and saves the return value."""

    def __init__(self, function, args, kwargs):
        super().__init__()
        self._call = (function, args, kwargs)
        self.return_value = None

    def run(self):
        function, args, kwargs = self._call
        self.return_value = function(*args, **kwargs)


class _DelayedFunctionWithReturnValue(threading.Timer):
    """Calls a function after a certain delay and saves the return value."""

    def __init__(self, seconds, function, args, kwargs):
        super().__init__(seconds, lambda: None)
        # Gives us a reliable timer while still allowing us to store the
        # return value
        self._call = (function, args, kwargs)
        self.return_value = None

    def run(self):
        super().run()
        function, args, kwargs = self._call
        self.return_value = function(*args, **kwargs)


def thread(function):
    """Run `function` in a separate thread."""
    def call(*args, **kwargs):
        thread = threading.Thread(
            target=function,
            args=args,
            kwargs=kwargs
        )
        thread.start()
    return call


def thread_join(function):
    """Run `function` in a separate thread and join the thread."""
    def call(*args, **kwargs):
        thread = threading.Thread(
            target=function,
            args=args,
            kwargs=kwargs
        )
        thread.start()
        thread.join()
    return call


def thread_daemon(function):
    """Run `function` in a separate daemon thread."""
    def call(*args, **kwargs):
        thread = threading.Thread(
            target=function,
            args=args,
            kwargs=kwargs
        )
        thread.daemon = True
        thread.start()
    return call


def thread_with_return_value(function):
    """Run `function` in a separate thread and return the return value.
    The function call will block due to the use of `join`.
    """
    def call(*args, **kwargs):
        thread = _FunctionThreadWithReturnValue(
            function=function,
            args=args,
            kwargs=kwargs
        )
        thread.start()
        thread.join()
        # Wait for the function to fall through to avoid
        # returning `None`
    return call


def delay(seconds):
    """Wait `seconds` seconds before calling `function`."""
    def wrap(function):
        def call(*args, **kwargs):
            function_timer = threading.Timer(
                seconds,
                function=function,
                args=args,
                kwargs=kwargs
            )
            function_timer.start()
        return call
    return wrap


def delay_join(seconds):
    """Wait `seconds` seconds before calling `function` and join the thread."""
    def wrap(function):
        def call(*args, **kwargs):
            function_timer = threading.Timer(
                seconds,
                function=function,
                args=args,
                kwargs=kwargs
            )
            function_timer.start()
            function_timer.join()
        return call
    return wrap


def delay_daemon(seconds):
    """Wait `seconds` seconds before calling `function` in a daemon thread."""
    def wrap(function):
        def call(*args, **kwargs):
            function_timer = threading.Timer(
                seconds,
                function=function,
                args=args,
                kwargs=kwargs
            )
            function_timer.daemon = True
            function_timer.start()
        return call
    return wrap


def delay_with_return_value(seconds):
    """Wait `seconds` seconds before returning the call to `function`."""
    def wrap(function):
        def call(*args, **kwargs):
            function_timer = _DelayedFunctionWithReturnValue(
                seconds=seconds,
                function=function,
                args=args,
                kwargs=kwargs
            )
            function_timer.start()
            function_timer.join()
            return function_timer.return_value
        return call
    return wrap

Example usage:

from sew import delay_with_return_value

numbers = [0, 1, 2, 3]


@delay_with_return_value(0.5)
def get_with_delay(index):
    """Wait half a second before returning numbers[index]."""
    return numbers[index]

for index in range(4):
    print(get_with_delay(index))
\$\endgroup\$

1 Answer 1

1
\$\begingroup\$

Can't really find any problems with the code. For example, trying to DRY (do not repeat yourself) further can only hurt the readability.

One thing maybe worth mentioning is adding parameters and return values to the functions. As this is a library, it's API's documentation is better be always up to date, so using some documentation generator directly from code (eg, Sphinx autodoc or see answers here) may help. Just decide on specific style for the documentation generator and add hints for it consistently. Your IDE / code editor may support one way or another already. Then you may provide more descriptive module level docstring, because it will not need to serve as an index of function names any more.

\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.