Home > Mobile >  Flask add custom background task
Flask add custom background task

Time:10-28

I am trying to create a flask application that has a background task that runs every 10 mins.

I found this module: https://medium.com/greedygame-engineering/an-elegant-way-to-run-periodic-tasks-in-python-61b7c477b679

So I added it to my project:

poetry add timeloop


I have created the hierachy using Flasky code examples(I don't have the book): https://github.com/miguelgrinberg/flasky


# app/__init__.py
  from flask import Flask
  from timeloop import Timeloop
  from app.utils.json_utils import JSON_Custom
  from flask_bootstrap import Bootstrap
  from config import config
  
  bootstrap = Bootstrap() : Bootstrap                                
    
    
    
  def create_app(config_name): -> Flask                                
      app = Flask(__name__) : Flask                                
      app.config.from_object(config[config_name])
      app.json_encoder = JSON_Custom
      config[config_name].init_app(app)
      bootstrap.init_app(app)
  
      tl = Timeloop() : Timeloop                                
      tl.start()
      with app.app_context():
          app.config['TIME_LOOP'] = tl : Timeloop                                
    
      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

As you can see I am trying to add the timeloop class instance to the context so I could reference it later on.

Now In my other file

#app/utils/connection_utils.py
@current_app.config['TIME_LOOP'].job(interval=timedelta(minutes=10))
def synchronize_applications():
  try:
    #code
  except:
    #code
  finally:
    current_app.config['TIME_LOOP'].stop()

Now the problem is that I have tried it many ways and each time I get a different error. With this setup I get

    raise RuntimeError(unbound_message) from None
RuntimeError: Working outside of application context.

This typically means that you attempted to use functionality that needed
the current application. To solve this, set up an application context
with app.app_context(). See the documentation for more information.

EDIT:

I also tried the following method.

#app/models.py
from timeloop import Timeloop

tl = Timeloop()

#app/__init__.py
def create_app(config_name):
  ...
  from app.models import tl
  tl.start()
  ...

#app/utils/connection_utils.py
from app.models import tl

@tl.job(interval=timedelta(minutes=10))
def synchronize_application():
  with current_app.app_context():
    ...

Now the flask APP starts, but it crashes when it encounters the TL.

[2022-10-26 12:54:11,626] [timeloop] [INFO] Starting Timeloop..
[2022-10-26 12:54:11,626] [timeloop] [INFO] Registered job <function synchronize_applications at 0x7fa1640984c0>
[2022-10-26 12:54:11,626] [timeloop] [INFO] Timeloop now started. Jobs will run based on the interval set
 * Serving Flask app 'main.py'
 * Debug mode: off
WARNING: This is a development server. Do not use it in a production deployment. Use a production WSGI server instead.
 * Running on http://127.0.0.1:5000
Press CTRL C to quit
Exception in thread Thread-1:
Traceback (most recent call last):
...
    raise RuntimeError(unbound_message) from None
RuntimeError: Working outside of application context.

This typically means that you attempted to use functionality that needed
the current application. To solve this, set up an application context
with app.app_context(). See the documentation for more information.

I think it is because it runs in a separate thread?

CodePudding user response:

In the end the solution was something like this:

I created a background tasks class that accepts the current app and the timeloop object who stores jobs.

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

 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)                                                                                                                                                                                                                                                             
                                                                                                                         
      # Pass the App and the timeloop object created to the backgroud tasks.                                             
      background_tasks = BackgroundTasks(app, timeloop) : BackgroundTasks                                  
      # Start the background tasks.                                                                                                                      
      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

class BackgroundTasks(threading.Thread):
      """Class that runs background tasks for a flask application.
      Args:
          threading.Thread: Creates a new thread.                                
      """
  
      def __init__(self, app: Flask, timeloop: Timeloop): -> None                                                                                                                                                                                                                         
          """Create a background tasks object that runs periodical tasks in the background of an flask application.
          Args:                                                      
              app: Flask application object.                                                               
              timeloop: Timeloop object.                                                                   
          """                                                                                                            
          super(BackgroundTasks, self).__init__()                                                                        
          self.app = app : Flask                                                                                                                                                                                                                                                          
          self.timeloop = timeloop : Timeloop                                                                                                                                                                                                                                             
          self.timeloop.add_job(self.synchronize_applications, interval=timedelta(minutes=10), 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_applicatons(self): -> None      
               #do stuff                                       

So in the end I didn't use the decorator, but I learned a lot about it and found this awesome module:

https://github.com/GrahamDumpleton/wrapt

https://www.youtube.com/watch?v=W7Rv-km3ZuA&t=32s

  • Related