Fix bug when run command celery (#569)

* fix bug when run command `celery`

* refactor get preload_options from base

* fix failed by flask8

* fix tests

* remove redundant

* fix typo

Author: lpthong90

djcelery/db.py

@@ -27,15 +27,15 @@ def commit_on_success(*args, **kwargs):
                 transaction.managed(True, *args, **kwargs)
                 try:
                     yield
-                except:
+                except Exception:
                     if transaction.is_dirty(*args, **kwargs):
                         transaction.rollback(*args, **kwargs)
                     raise
                 else:
                     if transaction.is_dirty(*args, **kwargs):
                         try:
                             transaction.commit(*args, **kwargs)
-                        except:
+                        except Exception:
                             transaction.rollback(*args, **kwargs)
                             raise
             finally:

djcelery/loaders.py

@@ -59,8 +59,8 @@ def read_configuration(self):
         self.configured = True
         # Default backend needs to be the database backend for backward
         # compatibility.
-        backend = (getattr(settings, 'CELERY_RESULT_BACKEND', None) or
-                   getattr(settings, 'CELERY_BACKEND', None))
+        backend = getattr(settings, 'CELERY_RESULT_BACKEND', None)
+        backend = backend or getattr(settings, 'CELERY_BACKEND', None)
         if not backend:
             settings.CELERY_RESULT_BACKEND = 'database'
         return DictAttribute(settings)

djcelery/management/base.py

@@ -35,15 +35,20 @@ def _init(self, *args, **kwargs):
                 __old__init__(self, *args, **kwargs)
                 self._thread_ident = _get_ident()
 
+            def _validate_thread(self):
+                is_valid = self.allow_thread_sharing
+                is_valid = is_valid or self._thread_ident == _get_ident()
+                return is_valid
+
             def _validate_thread_sharing(self):
-                if (not self.allow_thread_sharing and
-                        self._thread_ident != _get_ident()):
+                if not self.validate_thread():
                     raise DatabaseError(
                         DB_SHARED_THREAD % (
                             self.alias, self._thread_ident, _get_ident()),
                     )
 
             BaseDatabaseWrapper.__init__ = _init
+            BaseDatabaseWrapper.validate_thread = _validate_thread
             BaseDatabaseWrapper.validate_thread_sharing = \
                 _validate_thread_sharing
 
@@ -70,8 +75,8 @@ def add_arguments(self, parser):
                 option = {k: v
                           for k, v in opt.__dict__.items()
                           if v is not None}
-                flags = (option.get("_long_opts", []) +
-                         option.get("_short_opts", []))
+                flags = option.get("_long_opts", [])
+                flags += option.get("_short_opts", [])
                 if option.get('default') == ('NO', 'DEFAULT'):
                     option['default'] = None
                 if option.get("nargs") == 1:

djcelery/management/commands/celery.py

@@ -13,14 +13,8 @@ class Command(CeleryCommand):
     help = 'celery commands, see celery help'
     cc_options = CeleryCommand.options if CeleryCommand.options else []
     base_options = base.get_options() if base.get_options() else []
-    if hasattr(base, "preload_options"):
-        preload_options = base.preload_options if base.preload_options else []
-    else:
-        preload_options = []
-    preload_options = base.preload_options if base.preload_options else []
-    options = (cc_options +
-               base_options +
-               preload_options)
+    preload_options = getattr(base, 'preload_options', []) or []
+    options = cc_options + base_options + preload_options
 
     def run_from_argv(self, argv):
         argv = self.handle_default_options(argv)

djcelery/management/commands/celerybeat.py

@@ -15,10 +15,10 @@
 
 class Command(CeleryCommand):
     """Run the celery periodic task scheduler."""
-    options = (CeleryCommand.options +
-               beat.get_options() +
-               beat.preload_options)
     help = 'Old alias to the "celery beat" command.'
+    options = CeleryCommand.options
+    options += beat.get_options()
+    options += beat.preload_options
 
     def handle(self, *args, **options):
         beat.run(*args, **options)

djcelery/management/commands/celerycam.py

@@ -15,10 +15,10 @@
 
 class Command(CeleryCommand):
     """Run the celery curses event viewer."""
-    options = (CeleryCommand.options +
-               ev.get_options() +
-               ev.preload_options)
     help = 'Takes snapshots of the clusters state to the database.'
+    options = CeleryCommand.options
+    options += ev.get_options()
+    options += ev.preload_options
 
     def handle(self, *args, **options):
         """Handle the management command."""

djcelery/management/commands/celeryd.py

@@ -16,9 +16,9 @@
 class Command(CeleryCommand):
     """Run the celery daemon."""
     help = 'Old alias to the "celery worker" command.'
-    options = (CeleryCommand.options +
-               worker.get_options() +
-               worker.preload_options)
+    options = CeleryCommand.options
+    options += worker.get_options()
+    options += worker.preload_options
 
     def handle(self, *args, **options):
         worker.check_args(args)

djcelery/management/commands/celerymon.py

@@ -30,9 +30,9 @@
 
 class Command(CeleryCommand):
     """Run the celery monitor."""
-    options = (CeleryCommand.options +
-               (mon and mon.get_options() + mon.preload_options or ()))
     help = 'Run the celery monitor'
+    options = CeleryCommand.options
+    options += (mon and mon.get_options() + mon.preload_options or ())
 
     def handle(self, *args, **options):
         """Handle the management command."""

djcelery/management/commands/djcelerymon.py

@@ -32,9 +32,10 @@ def run(self):
 class Command(CeleryCommand):
     """Run the celery curses event viewer."""
     args = '[optional port number, or ipaddr:port]'
-    options = (runserver.Command.option_list +
-               ev.get_options() +
-               ev.preload_options)
+    options = runserver.Command.option_list
+    options += ev.get_options()
+    options += ev.preload_options
+
     help = 'Starts Django Admin instance and celerycam in the same process.'
     # see http://code.djangoproject.com/changeset/13319.
     stdout, stderr = sys.stdout, sys.stderr

djcelery/schedulers.py

@@ -159,10 +159,11 @@ def __init__(self, *args, **kwargs):
         self._dirty = set()
         self._finalize = Finalize(self, self.sync, exitpriority=5)
         Scheduler.__init__(self, *args, **kwargs)
-        self.max_interval = (
-            kwargs.get('max_interval') or
-            self.app.conf.CELERYBEAT_MAX_LOOP_INTERVAL or
-            DEFAULT_MAX_INTERVAL)
+        self.max_interval = kwargs.get('max_interval')
+        if self.max_interval is None:
+            self.max_interval = self.app.conf.CELERYBEAT_MAX_LOOP_INTERVAL
+        if self.max_interval is None:
+            self.max_interval = DEFAULT_MAX_INTERVAL
 
     def setup_schedule(self):
         self.install_default_entries(self.schedule)