Mercurial > kallithea

changeset 8830:853717af31d3

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
celery: let async tasks choose at runtime if they should use immediate execution or dispatch to the Celery worker Make it completely safe to use task annotation at import time, before global config has been set.
author Mads Kiilerich <mads@kiilerich.com>
date Fri, 01 Jan 2021 18:04:16 +0100
parents 01cb988e82a5
children 1f92cded1bd2
files kallithea/lib/celerylib/__init__.py
diffstat 1 files changed, 17 insertions(+), 16 deletions(-) [+]
line wrap: on
line diff
--- a/kallithea/lib/celerylib/__init__.py	Fri Jan 01 18:04:16 2021 +0100
+++ b/kallithea/lib/celerylib/__init__.py	Fri Jan 01 18:04:16 2021 +0100
@@ -43,29 +43,30 @@
 
 
 def task(f_org):
-    """Wrapper of celery.task.task, running async if CELERY_APP
+    """Wrapper of celery.task.task, run at import time, before kallithea.CONFIG has been set, and before kallithea.CELERY_APP has been configured.
     """
 
-    if asbool(kallithea.CONFIG.get('use_celery')):
-        def f_async(*args, **kwargs):
-            log.info('executing %s task', f_org.__name__)
-            try:
-                f_org(*args, **kwargs)
-            finally:
-                meta.Session.remove()  # prevent reuse of auto created db sessions
-                log.info('executed %s task', f_org.__name__)
-        runner = kallithea.CELERY_APP.task(name=f_org.__name__, ignore_result=True)(f_async)
+    def f_async(*args, **kwargs):
+        log.info('executing async task %s', f_org.__name__)
+        try:
+            f_org(*args, **kwargs)
+        finally:
+            meta.Session.remove()  # prevent reuse of auto created db sessions
+            log.info('executed async task %s', f_org.__name__)
 
-        def f_wrapped(*args, **kwargs):
+    runner = kallithea.CELERY_APP.task(name=f_org.__name__, ignore_result=True)(f_async)
+
+    def f_wrapped(*args, **kwargs):
+        if asbool(kallithea.CONFIG.get('use_celery')):
             t = runner.apply_async(args=args, kwargs=kwargs)
-            log.info('executing task %s in async mode - id %s', f_org, t.task_id)
-    else:
-        def f_wrapped(*args, **kwargs):
-            log.info('executing task %s in sync', f_org.__name__)
+            log.info('executing async task %s - id %s', f_org, t.task_id)
+        else:
+            # invoke f_org directly, without the meta.Session.remove in f_async
+            log.info('executing sync task %s', f_org.__name__)
             try:
                 f_org(*args, **kwargs)
             except Exception as e:
-                log.error('exception executing sync task %s in sync: %r', f_org.__name__, e)
+                log.error('exception executing sync task %s: %r', f_org.__name__, e)
                 raise # TODO: report errors differently ... and consistently between sync and async
 
     return f_wrapped