CODEWITHSUNDEEP
pythondecorator
rj487
rj487

Reputation: 4634

Use Python decorator to handle try and exception

How to pass try function and exception handler function in decorator way

For example, this is the WCF connection.

def wcf():
    def send_request(result):
        # establish connection...
        result["response"] = True

    def handle_execption(e, result):
        # logger something
        result["error"] = str(e)
    result = {}
    try:
        send_request(result)
    except Exception as e:
        handle_execption(e, result)

Now, I want to add a retry mechanism to this connection. What is the best way to achieve that, I have multiple connection ways(like REST, SOAP WCF, etc)? In general, they share the same pattern and all have the send_request and handle_execption.

The hard-code one is like the following, but it's quite silly to add the same logic to every protocol.

for attempt in range(0, 3):
    sleep_seconds = attempt ** 2
    sleep(sleep_seconds)

    try:
        send_request(result)
        break
    except Exception as e:
        handle_execption(e, result)

Upvotes: 3

Views: 1357

Answers (2)

water_ghosts
water_ghosts

Reputation: 736

If you always handle exceptions in the same way, you can write a decorator that hardcodes that behavior and handles the retrying logic:

def retry_decorator(base_function):
    def new_function(*args, **kwargs):  # This allows you to decorate functions without worrying about what arguments they take
        for attempt in range(3):
            sleep_seconds = attempt ** 2
            sleep(sleep_seconds)

            try:
                return base_function(*args, **kwargs)  # base_function is whatever function this decorator is applied to
            except Exception as e:
                print(e)  # Replace this with whatever you want, as long as it's the same for all possible base_functions

    return new_function


@retry_decorator  # Replaces send_request() with retry_decorator(send_request)()
def send_request(result):
    result["response"] = True

If you want to use different exception handling logic for the different connections, things get a bit more complicated:

def retry_decorator_2(exception_handle_function):
    def new_decorator(base_function):
        def new_function(*args, **kwargs):
            for attempt in range(3):
                sleep_seconds = attempt ** 2
                sleep(sleep_seconds)

                try:
                    return base_function(*args, **kwargs)
                except Exception as e:
                    exception_handle_function(e)

        return new_function

    return new_decorator


@retry_decorator_2(print)  # Replace 'print' with an exception handling function
def send_request(result):
    result["response"] = True

In this case, retry_decorator_2(print) creates a new decorator (which stores print as the exception_handle_function), then applies that decorator to the send_request function.

If you need the exception_handle_function to take arguments, you can pass them as part of the decorator:

def retry_decorator_3(exception_handle_function, *exception_function_args, **exception_function_kwargs):
    def new_decorator(base_function):
        def new_function(*args, **kwargs):
            for attempt in range(3):
                sleep_seconds = attempt
                sleep(sleep_seconds)

                try:
                    return base_function(*args, **kwargs)
                except Exception as e:
                    exception_handle_function(e, *exception_function_args, **exception_function_kwargs)

        return new_function

    return new_decorator


@retry_decorator_3(print, 1, 2, 3)  # 1, 2, 3 become arguments for the print function, or whatever you replace it with
def send_request(result):
    result["response"] = True

Upvotes: 2

md2perpe
md2perpe

Reputation: 3071

Why decorator? I think that it's just as clean with

def wcf():
    def send_request(result):
        # establish connection...
        result["response"] = True

    def handle_exception(e, result):
        # logger something
        result["error"] = str(e)

    return retry(send_request, handle_exception)


def retry(send_request, handle_exception):
    result = {}

    for attempt in range(0, 3):
        sleep(attempt ** 2)

        try:
            send_request(result)
            return result

        except Exception as e:
            return handle_execption(e, result)

Upvotes: 0

Related Questions