Search code examples
pythonpymongopython-decorators

How to make retry decorator for queries that shows string error as parameter


Im trying to make queries in my script using a retry decorator (I know there is a retry library, but I want know why im with this error), but I get an error.. I tried to change result = func() to result = func, but still with error. Can someone help?

import sys
import logging
from pymongo import MongoClient
import time


def retry(**kwargs):
    def wrapper(func):
        number_of_retries = 1
        while True:
            try:
                result = func()
                return result
            except Exception as e:
                logging.warning("Error {}. Attempt {}, error: {}".format(kwargs['error'], number_of_retries, e))
                number_of_retries += 1
                time.sleep(1)
                if number_of_retries > 5:
                    print("More than 5 attempts, closing the program")
                    sys.exit(1)
    return wrapper


@retry(error="connection with mongodb")
def mongo_connect():
    """" Function to connect with mongodb """
    client = MongoClient()
    client.server_info()
    db = client.test
    print(type(db))
    return db
db = mongo_connect()

The error:

raise TypeError("'Database' object is not callable. If you meant to "
TypeError: 'Database' object is not callable. If you meant to call the 'test' method on a 'MongoClient' object it is failing because no such method exists.

Solution

  • If you want to pass kwargs to decorators you need to do something like this (Decorators with parameters). To quote one of the answer's there:

    @decorator_with_args(arg)
    def foo(*args, **kwargs):
        pass
    

    Translates to:

    foo = decorator_with_args(arg)(foo)
    

    Simple decorators typically look like this:

    def decorator(func): 
        def wrapper(*args, **kwargs):
            returned_value = func(*args, **kwargs) 
            return returned_value 
        return wrapper 
    

    Here's a working demo for the behavior you want:

    import time
    import logging
    from functools import wraps
    
    def retry(**fkwargs):
        def _decorator(func):
            @wraps(func)
            def wrapper(*args, **kwargs):
                for attempt in range(1, 6):
                    try:
                        return func(*args, **kwargs)
                    except Exception as e:
                        logging.warning("Error: {}, Attempt: {}, "
                                        "Python Error: {}".format(fkwargs['error'], attempt, e))
                        time.sleep(1)
                print("More than 5 attempts, closing the program")
                # sys.exit(1)
            return wrapper
        return _decorator
    
    @retry(error="My custom error")
    def mock():
        raise ValueError("Connection Error")
    
    >>> mock()
    WARNING:root:Error: My custom error, Attempt: 1, Python Error: Connection Error
    WARNING:root:Error: My custom error, Attempt: 2, Python Error: Connection Error
    WARNING:root:Error: My custom error, Attempt: 3, Python Error: Connection Error
    WARNING:root:Error: My custom error, Attempt: 4, Python Error: Connection Error
    WARNING:root:Error: My custom error, Attempt: 5, Python Error: Connection Error
    More than 5 attempts, closing the program