Using Huey with Flask using application factory. For configuration, follow the djhuey format

app.py

# Due to the flexible nature of Flask. This might be __init__.py
from .extensions import huey

def create_app():
    app = Flask(__name__)

#     add your configurations
#     app.config.from_object("settings.local")

    huey.init_app(app)

    with app.app_context():
#       blueprints are registered here.

    return app

# So you just add a tasks.py to your project dir or blueprints folder
# Run the consumer using `flask run_huey`
# Viola!!!!!!!!!!!

cli.py

# This can either be located in yout project root directory or in a `core` blueprint(bp).
# This cli command lacks the use of flag. This was not necessary for my use case since I defined all in my config.
import logging

from flask import current_app
from huey.consumer_options import ConsumerConfig

from .utils import autodiscover_tasks


@current_app.cli.command("run_huey")
def run_huey():
    """Command to run huey consumer"""
    HUEY = current_app.extensions["huey"]

    consumer_options = {}

    try:
        if isinstance(current_app.config["HUEY"], dict):
            consumer_options.update(current_app.config["HUEY"].get("consumer", {}))
    except AttributeError:
        pass

    consumer_options.setdefault("verbose", consumer_options.pop("huey_verbose", None))

    # autodiscover every `tasks.py` module in every blueprints and project directory
    autodiscover_tasks()

    logger = logging.getLogger("huey")

    config = ConsumerConfig(**consumer_options)
    config.validate()

    if not logger.handlers:
        config.setup_logger(logger)

    consumer = HUEY.create_consumer(**config.values)
    consumer.run()

extensions.py

# Here all extensions are defined i.e
# db = SQLAlchemy()
# csrf = CSRFProtect()

class Huey:
    __app = None
    __instance = None

    def __init__(self, app=None):
        if app is not None:
            self.init_app(app)

    def init_app(self, app):
        self.__app = app

        _config = app.config.get("HUEY")

        if not _config:
            raise Exception("Huey requires a configuration in `app.config`")

        if not isinstance(_config, dict):
            raise Exception("Huey expects a dictionary as its configuration")

        _huey_instance = self.init_huey(_config)

        app.extensions["huey"] = _huey_instance
        self.__instance = _huey_instance

    def init_huey(self, config):
        _huey = RedisHuey(
            name=config.get("name", self.__app.name),
            url=config.get("url", None),
            result_store=config.get("result_store", False),
            events=config.get("events", True),
            store_none=config.get("store_none", False),
            always_eager=config.get("always_eager", False),
            store_errors=config.get("store_errors", True),
            blocking=config.get("blocking", False),
            read_timeout=config.get("read_timeout", 1),
        )
        return _huey

    def __getattr__(self, name):
        return getattr(self.__instance, name, None)
      

huey = Huey()

utils.py

def autodiscover_tasks():
    for key, value in current_app.blueprints.items():
        try:
            import_module(".tasks", value.import_name)

        except ModuleNotFoundError as err:
            pass

    # try importing from project root
    # if your create_app() is located in __init__.py, use the commented version
    # root_path = current_app.import_name
    root_path = ".".join(current_app.import_name.split(".")[:-1])
    try:
        import_module(".tasks", root_path)
    except ModuleNotFoundError as error:
        pass

say i wanted to use flask-sqlalchemy instance inside of a huey task where would i init the instance, would i create a new instance for my huey flask app something like:

db_huey = SQLAlchemy()

def huey_create_app(config, **kwargs):
    # create app
    huey_app = Flask("huey")
    ....
    
    # addconfig
    
    
    db_huey.init_app(huey_app )

something like this?. I was thinking importing the existing db instance from the original flask app and doing "init_app" on it inside huey's app factory might be dangerous

I've seen a lot of examples create another instance of the flask app inside each task, but I was able to get it working by passing the existing context that comes with click/the flask cli that the huey consumer is run in. Here are two methods:

from flask import current_app
from . import db

huey = current_app.huey
ctx = current_app.app_context()

@huey.context_task(ctx)
def some_task(some_args):
    current_app.logger.info(f'Doing task on {some_args}')
    result = db.session.execute() # [...etc]

@huey.periodic_task(crontab(minute='*'))
def minute_task():
    with ctx: # or ctx.push()
        current_app.logger.info('Peroidic task running...')

Where db is SQLAlchemy() which was init earlier in the app factory as normal. To match the original example current_app.huey would probably be current_app.extensions["huey"]