What is the best option for a (Python 3) task queue on Windows now that Celery 4 has dropped Windows support?
I run Flask with Huey on Windows without any issues, admittedly only for development and testing. For production I use Flask/Huey on Linux servers. Both with the Redis back-end, Flask 0.12 and Huey 1.2.0 .
I use the factory pattern to create a specialised "cut down" version of a Flask app for specific use by Huey tasks. This version doesn't load blueprints or configure Flask-Admin as these aren't required in the Huey tasks.
Example code of __init__.py
in app folder. App
is a class extending from Flask
:
def create_app(settings_override=None): app = App('app') if settings_override: app.config.from_object(settings_override) else: app.config.from_object(os.environ['APP_SETTINGS']) from .ext import configure_extensions configure_extensions(app, admin, load_modules=True) # REST import rest.api_v1 app.register_blueprint(api_v1_bp, url_prefix='/api/v1') # ... and more suffdef create_huey_app(): app = App('huey app') app.config.from_object(os.environ['APP_SETTINGS']) from .ext import configure_extensions configure_extensions(app, admin=None, load_modules=False) return app
The idea of configure_extensions
is taken from Quokka CMS. Examine its app
__init__.py
and its extensions module to see how this is implemented. Note also how this project too creates a specific app (create_celery_app
) for use with the Celery task queue.
Example of tasks.py
. Note the use of with app.app_context():
to create the Flask context. Now my functions have access to the extensions such as Flask-Mail, Flask-SqlAlchemy (db, models) etc.
@huey.task()def generate_transaction_documents_and_email(transaction_id): app = create_huey_app() with app.app_context(): reports.generate_transaction_documents_and_email(transaction_id)@huey.task()def send_email(subject, recipients, text_body, html_body, attachments=[], cc=[]): app = create_huey_app() with app.app_context(): emails.send_email(subject, recipients, text_body, html_body, attachments, cc)@huey.periodic_task(crontab(minute='30'))def synchronize_mailing_list(): app = create_huey_app() if app.config['CREATESEND_SYNCHRONIZE']: _list_name = app.config['CREATESEND_LIST'] with app.app_context(): sync_delete_ar_subscribers(_list_name) sync_add_ar_subscribers(_list_name)