我在Windows上与Huey一起运行Flask时没有任何问题,诚然仅用于开发和测试。对于生产,我在Linux服务器上使用Flask /
Huey。两者都具有Redis后端,分别是Flask 0.12和Huey 1.2.0。
我使用工厂模式来创建Flask应用程序的专用“缩减版本”,以供Huey任务特定使用。此版本不会加载蓝图或配置Flask-
Admin,因为Huey任务不需要这些。
__init__.py应用程序文件夹中的示例代码。
App是从扩展的类
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
的想法
configure_extensions来自Quokka
CMS。检查其
app
__init__.py及其扩展模块,以了解如何实现。还请注意,该项目还如何创建特定的应用程序(
create_celery_app),以与Celery任务队列一起使用。
的示例
tasks.py。请注意使用
with app.app_context():来创建Flask上下文。现在,我的函数可以访问诸如Flask-
Mail,Flask-SqlAlchemy(数据库,模型)等扩展。
@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)
欢迎分享,转载请注明来源:内存溢出
评论列表(0条)