In Miguel Grinberg's Flask Mega-Tutorial, chapter 22, he creates a generic launch_task method as a part of the User object model. His tutorial uses rq and I am trying to do something similar with celery. Here's the relevant code from his tutorial: class User(UserMixin, db.Model): # ... def launch_task(self, name, description, *args, **kwargs): rq_job = current_app.task_queue.enqueue(f'app.tasks.{name}', self.id, *args, **kwargs) task = Task(id=rq_job.get_id(), name=name, description=description, user=self) db.session.add(task) return task And here's an example of how it's called: @bp.route('/export_posts') @login_required def export_posts(): if current_user.get_task_in_progress('export_posts'): flash(_('An export task is currently in progress')) else: current_user.launch_task('export_posts', _('Exporting posts...')) db.session.commit() return redirect(url_for('main.user', username=current_user.username)) Where export_posts is a function defined in a tasks.py file: def export_posts(user_id): try: # read user posts from database # send email with data to user except Exception: # handle unexpected errors finally: # handle clean up How would you convert the launch_task method to use celery instead of rq? Continue reading...
