Wrappers

unique(max_retries=10, wait=0, alt_return='-no_alt_return-', exception=<class 'Exception'>, error_text=None)

Wrapper. Makes sure the function’s return value has not been returned before or else it run with the same inputs again.

Message format options: {func} {args} {kwargs}

Parameters:
  • max_retries – int of number of retries to attempt before failing
  • wait – float of seconds to wait between each try, defaults to 0
  • exception – Exception type of raise
  • error_text – text of the exception
  • alt_return – if specified, an exception is not raised on failure, instead the provided value of any type of will be returned
time_it(log=None, message=None, append=None)

Wrapper. Time the amount of time it takes the execution of the function and print it.

If log is true, make sure to set the logging level of ‘reusables’ to INFO level or lower.

import time
import reusables

reusables.add_stream_handler('reusables')

@reusables.time_it(log=True, message="{seconds:.2f} seconds")
def test_time(length):
    time.sleep(length)
    return "slept {0}".format(length)

result = test_time(5)
# 2016-11-09 16:59:39,935 - reusables.wrappers  INFO      5.01 seconds

print(result)
# slept 5

Message format options: {func} {seconds} {args} {kwargs}

Parameters:
  • log – log as INFO level instead of printing
  • message – string to format with total time as the only input
  • append – list to append item too
catch_it(exceptions=(<class 'Exception'>, ), default=None, handler=None)

If the function encounters an exception, catch it, and return the specified default or sent to a handler function instead.

def handle_error(exception, func, *args, **kwargs):
    print(f"{func.__name__} raised {exception} when called with {args}")

@reusables.catch_it(handler=err_func)
def will_raise(message="Hello")
    raise Exception(message)
Parameters:
  • exceptions – tuple of exceptions to catch
  • default – what to return if the exception is caught
  • handler – function to send exception, func, *args and **kwargs
log_exception(log='reusables', message=None, exceptions=(<class 'Exception'>, ), level=40, show_traceback=True)

Wrapper. Log the traceback to any exceptions raised. Possible to raise custom exception.

@reusables.log_exception()
def test():
    raise Exception("Bad")

# 2016-12-26 12:38:01,381 - reusables   ERROR  Exception in test - Bad
# Traceback (most recent call last):
#     File "<input>", line 1, in <module>
#     File "reusables\wrappers.py", line 200, in wrapper
#     raise err
# Exception: Bad

Message format options: {func} {err} {args} {kwargs}

Parameters:
  • exceptions – types of exceptions to catch
  • log – log name to use
  • message – message to use in log
  • level – logging level
  • show_traceback – include full traceback or just error message
retry_it(exceptions=(<class 'Exception'>, ), tries=10, wait=0, handler=None, raised_exception=<class 'reusables.shared_variables.ReusablesError'>, raised_message=None)

Retry a function if an exception is raised, or if output_check returns False.

Message format options: {func} {args} {kwargs}

Parameters:
  • exceptions – tuple of exceptions to catch
  • tries – number of tries to retry the function
  • wait – time to wait between executions in seconds
  • handler – function to check if output is valid, must return bool
  • raised_exception – default is ReusablesError
  • raised_message – message to pass to raised exception
lock_it(lock=<unlocked _thread.lock object>)

Wrapper. Simple wrapper to make sure a function is only run once at a time.

Parameters:lock – Which lock to use, uses unique default
queue_it(queue=<queue.Queue object>, **put_args)

Wrapper. Instead of returning the result of the function, add it to a queue.

Parameters:queue – Queue to add result into