mirror of
https://github.com/GSA/notifications-api.git
synced 2025-12-21 07:51:13 -05:00
2622866622
they were always caught locally by celery's base handler, however, we weren't logging them ourselves, which meant it wouldn't be put into the json logs that are sent to cloudwatch.
29 lines
799 B
Python
29 lines
799 B
Python
from flask import current_app
|
|
from celery import Celery, Task
|
|
|
|
|
|
class NotifyTask(Task):
|
|
abstract = True
|
|
|
|
def on_failure(self, exc, task_id, args, kwargs, einfo):
|
|
# ensure task will log exceptions to correct handlers
|
|
current_app.logger.exception('Celery task failed')
|
|
super().on_failure(exc, task_id, args, kwargs, einfo)
|
|
|
|
def __call__(self, *args, **kwargs):
|
|
# ensure task has flask context to access config, logger, etc
|
|
with current_app.app_context():
|
|
return super().__call__(*args, **kwargs)
|
|
|
|
|
|
class NotifyCelery(Celery):
|
|
|
|
def init_app(self, app):
|
|
super().__init__(
|
|
app.import_name,
|
|
broker=app.config['BROKER_URL'],
|
|
task_cls=NotifyTask,
|
|
)
|
|
|
|
self.conf.update(app.config)
|