Creating a class that has a decorator that interacts with class object

[TafkaMax]

Hi

This post isn't about a specific bug rather I need advice.

I am trying to use: https://github.com/sankalpjonn/timeloop

Or rather a version of it and developing a new version from it.

I am trying to create a job factory inside and it is actually really difficult.

---

First I create the object that stores the necessary jobs.

#app/models.py
from timeloop import Timeloop
timeloop = Timeloop()

Below is my create_app factory for flask. Here I import the created Timeloop object and send it to another thread.

#app/__init__.py
from app.models import timeloop

bootstrap = Bootstrap() : Bootstrap                                

  def create_app(config_name): -> Flask                                                        
      app = Flask(__name__) : Flask                                
      # Import config
      app.config.from_object(config[config_name])
      app.json_encoder = JSON_Custom
      config[config_name].init_app(app)
      # Add bootstrap
      bootstrap.init_app(app)

      background_tasks = SynchronizeThreadedTask(app, timeloop) : SynchronizeThreadedTask                                
      background_tasks.start()

      from .utils import utils as utils_blueprint
      app.register_blueprint(utils_blueprint)

      from .api import api as api_blueprint
      app.register_blueprint(api_blueprint, url_prefix='/api/v1')                                                          

      from .passwd import passwd as passwd_blueprint                                                                       
      app.register_blueprint(passwd_blueprint)

      return app

Here is the secondary thread that should run in flask background, but gets all the configuration/context of the flask app.

#app/utils/utils.py
class SynchronizeThreadedTask(threading.Thread):

      def __init__(self, app: Flask, timeloop: Timeloop): -> None                                                                                                         
          super(SynchronizeThreadedTask, self).__init__()
          self.app = app : Flask                                      
          self.timeloop = timeloop : Timeloop                                                     

      def run(self): -> None                                      
          with self.app.app_context():
              print('Running threaded task - synchornization')
              self.timeloop.start()                                  

     @timeloop(interval=timedelta(minutes=1), stop_on_exception=False)
     def synchronize_applications(self):

And below is my attempt at using wrapt.

 class Timeloop(object):
      def __init__(self, logger = None): -> None                                
          """Create Timeloop object that control all job.

          Keyword Arguments:
              logger {logging} -- If you have already a logger you can set with 
                  this the logger of Timeloop. (default: {None})
          """                                          
          self._jobs = {"to_run": [], "active": {}} : dict[str, Unknown]                                
          self._block = False : Literal[False]                                
          self._already_started = False : Literal[False]                                
          if not logger:
              ch = logging.StreamHandler(sys.stdout) : StreamHandler[TextIO]                                
              ch.setLevel(logging.INFO)
              ch.setFormatter(logging.Formatter('[%(asctime)s] [%(name)s] [%(levelname)s] %(message)s'))
              logger = logging.getLogger('timeloop')
              logger.addHandler(ch)
              logger.setLevel(logging.INFO)
          self._logger = logger : Logger | Unknown                                

      @wrapt.decorator
>>    def __call__(self, wrapped, instance, args, kwargs): -> ((wrapped: Unknown, instance: Unknown, args: Unknown, kwargs: Unknown)                                                                                                                                                      
          def wrapper(wrapped, instance, args, kwargs): -> None                                                                                                                                                                                                                           
              print(self)
              print(wrapped)
              print(instance)
              print(args)
              print(kwargs)
          return wrapper

Running this creates the following error

  File "app/utils/utils.py", line 276, in SynchronizeThreadedTask
    @timeloop(interval=timedelta(minutes=1), stop_on_exception=False)
  File "python3.10/site-packages/wrapt/decorators.py", line 308, in _wrapper
    target_wrapped = args[0]
IndexError: tuple index out of range

---

In the documentation there isn't much said about my use case.

Class instance that contains a decorator that is used on a another classes instance function.

[GrahamDumpleton]

If you are trying to mimic:

    def job(self, interval):
        def decorator(f):
            self._add_job(f, interval)
            return f
        return decorator

but where you are using the fact that the class instance can be callable, it would be:

    def __call__(self, interval):
        def decorator(f):
            self._add_job(f, interval)
            return f
        return decorator

I am not even sure a wrapt type decorator is even needed as the function called decorator here isn't really a decorator in the sense of fulfilling a role of providing a wrapper that is called each time the function is called, instead it is performing a one off action when the code file is first parsed and processed, with the original function still being returned.

Naming wise it is possibly better written as the following so intent is clearer and the name decorator being used doesn't confuse things.

    def __call__(self, interval):
        def add_job(f):
            self._add_job(f, interval)
            return f
        return add_job

[TafkaMax]

Hi

Thanks for the insight.

In the end I didn't use the decorator function and opted for another way to do this.

I renamed the SynchronizeThreadedTask to BackgroundTasks and this seems to work. I am using another version of timeloop

https://github.com/Ruggiero-Santo/timeloop

class BackgroundTasks(threading.Thread):                                                                                                             
      """Class that runs background tasks for a flask application.                                                                                     
      """                                                                                                                                              

      def __init__(self, app: Flask, timeloop: Timeloop): -> None                                                                                      
          super(BackgroundTasks, self).__init__()
          self.app = app : Flask                                                                                                                                                                      
          self.timeloop = timeloop : Timeloop                                                                                                                                                         
          self.timeloop.add_job(self.synchronize_applications, interval=timedelta(minutes=1), exception=False)                                    

      def run(self): -> None                                
          # Use the current application context and start the timeloop service.                                 
          with self.app.app_context():
              self.timeloop.start()

      def synchronize_applications(self): -> bool                                

---

In the end the current solution works, but I will try to look at ways to incorporate using decorators instead of writing inside the init function.

Nevertheless this whole journey at decorators has broadened my knowledge on the topic and it's usages and how it is commonly misunderstood. Thanks for the wrapt package :) .