Django-celery-beat: Does django-celery-beat handle django database routing?

[wiretiger]

I have multiple databases setup in my django instance. The 'default' database is not defined.

DATABASES = {
'default': {
},
'jobdb': {
'ENGINE': 'django.db.backends.postgresql',
'NAME': 'jobdb',
'USER': 'user',
'PASSWORD': 'XXXXXXXXXXXX',
},
'appdb': {
'ENGINE': 'django.db.backends.postgresql',
'NAME': 'appdb',
'USER': 'user',
'PASSWORD': 'XXXXXXXXXXXX',
}

I then have a very simple database router that returns the correct database based on the 'app' that model/object is a member of.

APP_TO_DATABASE = {
'myapp' : 'appdb',
'django_celery_beat': 'jobdb',
}
class AppDbRouter(object):
"""
A router to control all database operations on models using a simple
dictionary mapping.
"""
def db_for_read(self, model, *hints):
""" dictionary mapping for read database """
try:
return APP_TO_DATABASE[model._meta.app_label]
except KeyError:
return None
def db_for_write(self, model, *hints): # pylint: disable=unused-argument
""" dictionary mapping for write database """
try:
return APP_TO_DATABASE[model._meta.app_label]
except KeyError:
return None
def allow_relation(self, obj1, obj2, *hints): # pylint: disable=unused-argument
"""
Allow relations if a models are in the same db
"""
try:
return (APP_TO_DATABASE[obj1._meta.app_label]
== APP_TO_DATABASE[obj2._meta.app_label])
except KeyError:
return None
def allow_migrate(self, db, app_label, model_name=None, *hints): # pylint: disable=unused-argument
"""
Make sure the app only appears in the correct database.

When I run celery beat with the DatabaseScheduler I get errors because it is trying to use the 'default' database.
celery -A celery_beat beat --scheduler=django_celery_beat.schedulers.DatabaseScheduler
I get the following error:

[2018-05-31 15:56:51,578: CRITICAL/MainProcess] beat raised exception : ImproperlyConfigured('settings.DATABASES is improperly configured. Please supply the ENGINE value. Check settings documentation for more details.',)

Looking at the code, it does not appear as if the DatabaseScheduler ever tries to figure out which database to use by calling the database router.

Using the PeriodTask, CrontabSchedule objects to load up scheduled tasks works fine and are added to the correct DB. Just not the actual beat scheduler.

[wiretiger]

I was eventually able to get around this issue by sub-classing django_celery_beat.schedulers.DatabaseScheduler and overriding thesync and schedule_change functions to add a single line to get the correct db from the database router and then passing that to the django transaction functions.

from django_celery_beat.schedulers import DatabaseScheduler
class MyDatabaseScheduler(DatabaseScheduler):

    def sync(self):
        info('Writing entries...')
        _tried = set()
        db = router.db_for_write(self.Model)
        try:
            close_old_connections()
            with transaction.atomic(using=db):
                while self._dirty:
                    try:
                        name = self._dirty.pop()
                        _tried.add(name)
                        self.schedule[name].save()
                    except (KeyError, ObjectDoesNotExist):
                        pass
        except DatabaseError as exc:
            # retry later
            self._dirty |= _tried
            logger.exception('Database error while sync: %r', exc)

    def schedule_changed(self):
        try:
            # If MySQL is running with transaction isolation level
            # REPEATABLE-READ (default), then we won't see changes done by
            # other transactions until the current transaction is
            # committed (Issue #41).
            db = router.db_for_write(self.Model)
            try:
                transaction.commit(using=db)
            except transaction.TransactionManagementError:
                pass  # not in transaction management.

            last, ts = self._last_timestamp, self.Changes.last_change()
        except DatabaseError as exc:
            logger.exception('Database gave error: %r', exc)
            return False
        try:
            if ts and ts > (last if last else ts):
                return True
        finally:
            self._last_timestamp = ts
        return False

[syurchenko]

Can you use models of celery beat in your code? may be there is bug in your router

[srikanth-chandaluri]

I am also facing the same issue. After adding default databases it is working fine( my default database don't have any tables )

[auvipy]

what's it's the update of this issue with 1.4.0?

[Allan-Nava]

T

I was eventually able to get around this issue by sub-classing django_celery_beat.schedulers.DatabaseScheduler and overriding thesync and schedule_change functions to add a single line to get the correct db from the database router and then passing that to the django transaction functions.

from django_celery_beat.schedulers import DatabaseScheduler
class MyDatabaseScheduler(DatabaseScheduler):

    def sync(self):
        info('Writing entries...')
        _tried = set()
        db = router.db_for_write(self.Model)
        try:
            close_old_connections()
            with transaction.atomic(using=db):
                while self._dirty:
                    try:
                        name = self._dirty.pop()
                        _tried.add(name)
                        self.schedule[name].save()
                    except (KeyError, ObjectDoesNotExist):
                        pass
        except DatabaseError as exc:
            # retry later
            self._dirty |= _tried
            logger.exception('Database error while sync: %r', exc)

    def schedule_changed(self):
        try:
            # If MySQL is running with transaction isolation level
            # REPEATABLE-READ (default), then we won't see changes done by
            # other transactions until the current transaction is
            # committed (Issue #41).
            db = router.db_for_write(self.Model)
            try:
                transaction.commit(using=db)
            except transaction.TransactionManagementError:
                pass  # not in transaction management.

            last, ts = self._last_timestamp, self.Changes.last_change()
        except DatabaseError as exc:
            logger.exception('Database gave error: %r', exc)
            return False
        try:
            if ts and ts > (last if last else ts):
                return True
        finally:
            self._last_timestamp = ts
        return False

This code it works for multidb and django celery beat?

[crucialfelix]

django celery beat does use the Django router. Since it's just making a db query using the Django ORM it would have to in less it explicitly did some hack to avoid using it.

Note the first line in my logs:

beat_1       | DEBUG db_for_read django_celery_beat periodictasks use: default
beat_1       | [0.003] SELECT typarray
beat_1       | FROM pg_type
beat_1       | WHERE typname = 'citext'
beat_1       |
beat_1       | [0.007] SELECT "django_celery_beat_periodictasks"."ident",
beat_1       |        "django_celery_beat_periodictasks"."last_update"
beat_1       | FROM "django_celery_beat_periodictasks"
beat_1       | WHERE "django_celery_beat_periodictasks"."ident" = 1