Correctly Setup Celery with Flask-Application Factory Pattern/Gunicorn/Nginx/Supervisor

Viewed 210

I have a task of updating every single row of a MySQL table but it's super slow. I rarely need to do it and only when I change something fundamental, but I thought this would be a great change to learn about multi threading. However all the examples and tutorials online go over some things and not others and I'm struggling to piece all the information together.
I know I need to make a celery process I just don't know if I'm doing it right. A lot of tutorials talk about dockerizing a redis environment without explaining how to do it so I thought I'd come here for some real human-to-human interaction to maybe help me feel less stupid about this.Here's my code so far
/website/__init__.py

from flask import Flask, appcontext_popped, render_template
from flask_sqlalchemy import SQLAlchemy
from flask_login import LoginManager, UserMixin, login_user, login_required, logout_user, current_user
from flask_migrate import Migrate
from flask_wtf import CSRFProtect
import logging

import celery

#Path Math
import sys
import os
from . import config

db:SQLAlchemy = SQLAlchemy()
migrate = Migrate()
csrf = CSRFProtect()
celery: celery.Celery

DB_NAME = "main"

def create_app(name):
    #Flask Instance
    app = Flask(__name__)
    app.config.from_object(config.ProdTestConfig)
    
    # logging stuff

    #Database
    db.init_app(app)
    migrate.init_app(app, db)
    csrf.init_app(app)
    global celery
    celery = make_celery(app)
    
    with app.app_context():
        db.create_all()

        # Models and Blueprints here

        from .helper_functions import migration_handling as mgh
        #where you will find the thing I need to run async
        app.before_first_request(mgh.run_back_check)

        # log manager stuff

    #error page handling

    return app


def make_celery(app):
    celery = celery.Celery(
        app.import_name,
        backend=app.config['CELERY_RESULT_BACKEND'],
        broker=app.config['CELERY_BROKER_URL']
    )
    celery.conf.update(app.config)

    class ContextTask(celery.Task):
        def __call__(self, *args, **kwargs):
            with app.app_context():
                return self.run(*args, **kwargs)

    celery.Task = ContextTask
    return celery

I've read some other ways seem to fit a bit better like using:
celery = Celery(__name__, broker=Config.CELERY_BROKER_URL, result_backend=Config.RESULT_BACKEND)

Then in create_app() they run celery.conf.update(app.config). The issue with this is that I don't know how to setup a redis server on my linode machine hosting the site and my personal windows machine. I have redis pip installed. This is how the function I'm trying to run async looks:

@celery.task(name='app.tasks.campaign_pay_out_process')
def campaign_pay_out_process():
    '''
    Process Every Campaigns Pay
    '''
    campaign: Campaigns
    for campaign in Campaigns.query.filter_by():
        campaign.process_pay()
    db.session.commit()
    current_app.logger.info('Done Campaign Pay Out Processing')

I'm running gunicorn off of supervisor because restarting is super easy and ridding my life of super long linux commands to start a process has been great. I know this is the command for celery: celery -A celery_worker.celery worker --pool=solo --loglevel=info and I'd love to know how to include that in my work flow. Here's my supervisor config:

[program:paymentwebapp]
directory=/home/sai/paymentWebApp
command=/home/sai/paymentWebApp/venv/bin/gunicorn --workers 1 --threads 3  wsgi:app
user=sai
autostart=true
autorestart=true
stopasgroup=true
killasgroup=true
stderr_logfile=/var/log/paymentwebapp/paymentwebapp.err.log
stdout_logfile=/var/log/paymentwebapp/paymentwebapp.out.log

Here's my flask config right now:

from os import environ, path
from dotenv import load_dotenv

DB_NAME = "main"

class Config:
    """Base config."""
    #SESSION_COOKIE_NAME = environ.get('SESSION_COOKIE_NAME')
    MAX_CONTENT_LENGTH = 16*1000*1000
    RECEIPT_FOLDER = '../uploads/receipts'
    IMPORT_FOLDER = 'uploads/imports'
    UPLOAD_FOLDER = 'uploads'
    EXPORT_FOLDER = '/uploads/exports'
    UPLOAD_EXTENSIONS = ['.jpg', '.png', '.pdf', '.csv', '.xls', '.xlsx']
    STATIC_FOLDER = 'static'
    TEMPLATES_FOLDER = 'templates'

class ProdConfig(Config):
    basedir = path.abspath(path.dirname(__file__))
    load_dotenv('/home/sai/.env')
    env_dict = dict(environ)
    FLASK_ENV = 'production'
    DEBUG = False
    TESTING = False
    SQLALCHEMY_DATABASE_URI = environ.get('PROD_DATABASE_URI')
    SECRET_KEY = environ.get('SECRET_KEY')
    SERVER_NAME = environ.get('SERVER_NAME')
    SESSION_COOKIE_SECURE = True
    WTF_CSRF_TIME_LIMIT = 600

    #Uploads


class DevConfig(Config):
    basedir = path.abspath(path.dirname(__file__))
    load_dotenv('C:\saiscripts\intercept_branch\Payment Web App Project\.env')
    env_dict = dict(environ)
    FLASK_ENV = 'development'
    DEBUG = True
    SQLALCHEMY_DATABASE_URI = environ.get('DEV_DATABASE_URI')
    SECRET_KEY = environ.get('SECRET_KEY')
    
class ProdTestConfig(DevConfig):
    '''
    Developer config settings but production database server
    '''
    SQLALCHEMY_DATABASE_URI = environ.get('PROD_DATABASE_URI')

if __name__ == '__main__':
    print(environ.get('SQLALCHEMY_DATABASE_URI'))

This is where I copied some code from a tutorial because I'm supposed to make a celery worker:

#!/usr/bin/env python
import os
#from app import create_app, celery
from website import create_app

app = create_app()
app.app_context().push()

from website import celery
0 Answers
Related