reloadable 0.1.4

Usage

The function my_func will run indefinitely until it stops raising exceptions, which will never happen in this case.

from reloadable import reloadable

@reloadable()
def my_func():
    raise Exception('Oops')

This module is useful when we want to run something forever, like a code that connects to a queue en fetches messages. Eventually it may disconnect and raise an error trying to fetch a message, so reloadable can retry connecting.

@reloadable()
def get_message():
    conn = Queue(host='...', password='...')

    while True:
        message = conn.fetch_message()
        # probably process message afterwards...

You can config a callback function that receives an exception, which will be called if it occurs.

def shit_happens(exception):
    logger.exception(exception)

@reloadable(exception_callback=shit_happens)
def dont_stop():
    raise Exception('Deal with it')

You can also wait some time before the next respawn

@reloadable(sleep_time=7)  # wait 7 seconds before running `get_message` after a failure
def get_message():
    # some code...

You can always stop reloadable with a KeyboardInterrupt exception (usually triggered by ^C, but not necessarily).

Another option is to configure the stop condition exception.

@reloadable(stop_condition_exception=ValueError)
def i_will_stop():
    raise ValueError('something went wrong')

Or you can define it globally, which will be used if local stop condition wasn’t defined

from reloadable import reloadable, configure

configure(stop_condition_exception=KeyError)

@reloadable()
def i_will_stop():
    raise KeyError('...')

You may also want to limit the number of times that the decorator should try rerun the function

from reloadable import reloadable

@reloadable(max_reloads=2)
def a_func():
    raise KeyError('...')

Alternatively you can disable the reloadable decorator via configuration, which is useful during unittests.

from reloadable import configure, reloadable

configure(enabled=False)

@reloadable()  # When disabled, it does nothing
def i_am_free():
    return '\o/'

Tests

python -m unittest -v tests

Installation

pip install reloadable