Django-celery-beat: Error with settings.USE_TZ=False

[kosmos342]

With celery 4.1 and without using timezone support in django I got

ValueError("MySQL backend does not support timezone-aware datetimes when USE_TZ is False.")

because in:
https://github.com/celery/django-celery-beat/blob/b6c0581756a7e47e356a5ca64c674142402f9237/django_celery_beat/schedulers.py#L87

https://github.com/celery/django-celery-beat/blob/b6c0581756a7e47e356a5ca64c674142402f9237/django_celery_beat/schedulers.py#L107

uses https://github.com/celery/celery/blob/5af199c94f3bf08dfaa93c36c83c2147b5f15639/celery/app/base.py#L871
which return timezone-aware datetime

[117111302]

I got this too

[kosmos342]

@117111302, for now I added this to avoid this error:

app = Celery(.....)
app.now = datetime.datetime.utcnow

[bluekingassistant]

I got this too

[jeremylv1992]

I got this too, 'use_tz=True' must be set??

[assertyang]

i think this s caused by celery-beat not support none UTC timezone well. my setting is below :
USE_TZ = False TIME_ZONE = 'Asia/Shanghai' USE_I18N = True USE_L10N = True CELERY_ENABLE_UTC = False CELERY_TIMEZONE = 'Asia/Shanghai'
as we know ,if timezone used in django ,when we save model , datetime object will have tzinfo instead of none.
i tried many ways ,and last have to modify my project celery.py from @kosmos342
`app = Celery('xxxxx')

app.now = datetime.now`

then new problem i met when try to modify task via django admin
[2018-03-16 14:12:23,012: WARNING/MainProcess] return now.tzinfo.localize(now.replace(tzinfo=None)) [2018-03-16 14:12:23,012: WARNING/MainProcess] AttributeError [2018-03-16 14:12:23,012: WARNING/MainProcess] : [2018-03-16 14:12:23,012: WARNING/MainProcess] 'NoneType' object has no attribute 'localize'

this caused by app.now return a datetime object without tzinfo

after test ,i changed celery-beat source code to support localtime
first： django_celery_beat.utils modify make_aware method， if celery_enable_utc setting is false
should change time to native time

`
def make_aware(value):

"""Force datatime to have timezone information."""

if getattr(settings, 'USE_TZ', False):

    # naive datetimes are assumed to be in UTC.

    if timezone.is_naive(value):

        value = timezone.make_aware(value, timezone.utc)

    # then convert to the Django configured timezone.

    default_tz = timezone.get_default_timezone()

    value = timezone.localtime(value, default_tz)

if not getattr(settings,'CELERY_ENABLE_UTC', True):

    if timezone.is_naive(value):

        value = value

    else:

        value = timezone.make_naive(value)

return value

`

and django_celery_beat.schedulers.py ,directly return a localtime

"""django_celery_beat.schedulers.py"""
`
def _default_now(self):

    now = self.app.now()

    if is_aware(now):

        return now.replace(tzinfo=None)  

    else:

        return now

`

[im-n1]

There is a lot of project which don't use timezones at all.

This bug makes celery pretty much unusable with MySQL. Is here any progress so far?

[zuhao]

@grafa The most painless solution I have found so far is to downgrade django-celery-beat to version 1.1.0. Everything should work fine, in the meanwhile just wait for upstream fixes.

[im-n1]

@kosmos342 solution works!

[pylixm]

The above issue was resolved using the 1.1.0 version. But when I modify the Interval of the task, I still report the above error!
Has anyone ever encountered it?

[cmberryau]

Using postgres with USE_TZ=False results in scheduler spamming task messages non stop once first message is sent

[zuoakang]

MySQL backend does not support timezone-aware datetimes when USE_TZ is False.

Who can solve this problem?

ENV

• python3.6.5
• kombu 4.1.0
• django-celery-beat 1.1.1
• celery 4.1.0

[light1988]

I also solved it downgrading to version 1.1.0 and and adding to _celery.py_ the following as @zuhao and @kosmos342 said:

app.now = datetime.datetime.now

[xiaohuazi123]

MySQL backend does not support timezone-aware datetimes when USE_TZ is False.

Who can solve this problem?

ENV

• python3.6.5
• kombu 4.1.0
• django-celery-beat 1.1.1
• celery 4.1.0

猪八戒网也做运管平台了

[xiaohuazi123]

"""Beat Scheduler Implementation."""
from __future__ import absolute_import, unicode_literals

import logging

from multiprocessing.util import Finalize

from celery import current_app
from celery import schedules
from celery.beat import Scheduler, ScheduleEntry
from celery.five import values, items
from celery.utils.encoding import safe_str, safe_repr
from celery.utils.log import get_logger
from kombu.utils.json import dumps, loads

from django.db import transaction, close_old_connections
from django.db.utils import DatabaseError
from django.core.exceptions import ObjectDoesNotExist
from django.utils.timezone import is_aware

from .models import (
PeriodicTask, PeriodicTasks,
CrontabSchedule, IntervalSchedule,
SolarSchedule,
)
from .utils import make_aware

try:
from celery.utils.time import is_naive
except ImportError: # pragma: no cover
from celery.utils.timeutils import is_naive # noqa

This scheduler must wake up more frequently than the

regular of 5 minutes because it needs to take external

changes to the schedule into account.

DEFAULT_MAX_INTERVAL = 5 # seconds

ADD_ENTRY_ERROR = """
Cannot add entry %r to database schedule: %r. Contents: %r
"""

logger = get_logger(__name__)
debug, info = logger.debug, logger.info

class ModelEntry(ScheduleEntry):
"""Scheduler entry taken from database row."""

model_schedules = (
    (schedules.crontab, CrontabSchedule, 'crontab'),
    (schedules.schedule, IntervalSchedule, 'interval'),
    (schedules.solar, SolarSchedule, 'solar'),
)
save_fields = ['last_run_at', 'total_run_count', 'no_changes']

def __init__(self, model, app=None):
    """Initialize the model entry."""
    self.app = app or current_app._get_current_object()
    self.name = model.name
    self.task = model.task
    try:
        self.schedule = model.schedule
    except model.DoesNotExist:
        logger.error(
            'Disabling schedule %s that was removed from database',
            self.name,
        )
        self._disable(model)
    try:
        self.args = loads(model.args or '[]')
        self.kwargs = loads(model.kwargs or '{}')
    except ValueError as exc:
        logger.exception(
            'Removing schedule %s for argument deseralization error: %r',
            self.name, exc,
        )
        self._disable(model)

    self.options = {
        'queue': model.queue,
        'exchange': model.exchange,
        'routing_key': model.routing_key,
        'expires': model.expires,
    }
    self.total_run_count = model.total_run_count
    self.model = model

    if not model.last_run_at:
        model.last_run_at = self._default_now()
    self.last_run_at = make_aware(model.last_run_at)

def _disable(self, model):
    model.no_changes = True
    model.enabled = False
    model.save()

def is_due(self):
    if not self.model.enabled:
        return False, 5.0   # 5 second delay for re-enable.
    return self.schedule.is_due(self.last_run_at)

def _default_now(self):
    now = self.app.now()
    # The PyTZ datetime must be localised for the Django-Celery-Beat
    # scheduler to work. Keep in mind that timezone arithmatic
    # with a localized timezone may be inaccurate.
    # return now.tzinfo.localize(now.replace(tzinfo=None))
    if is_aware(now):
        return now.replace(tzinfo=None)
    else:
        return now

def __next__(self):
    self.model.last_run_at = self.app.now()
    self.model.total_run_count += 1
    self.model.no_changes = True
    return self.__class__(self.model)
next = __next__  # for 2to3

def save(self):
    # Object may not be synchronized, so only
    # change the fields we care about.
    obj = type(self.model)._default_manager.get(pk=self.model.pk)
    for field in self.save_fields:
        setattr(obj, field, getattr(self.model, field))
    obj.save()

@classmethod
def to_model_schedule(cls, schedule):
    for schedule_type, model_type, model_field in cls.model_schedules:
        schedule = schedules.maybe_schedule(schedule)
        if isinstance(schedule, schedule_type):
            model_schedule = model_type.from_schedule(schedule)
            model_schedule.save()
            return model_schedule, model_field
    raise ValueError(
        'Cannot convert schedule type {0!r} to model'.format(schedule))

@classmethod
def from_entry(cls, name, app=None, **entry):
    return cls(PeriodicTask._default_manager.update_or_create(
        name=name, defaults=cls._unpack_fields(**entry),
    ), app=app)

@classmethod
def _unpack_fields(cls, schedule,
                   args=None, kwargs=None, relative=None, options=None,
                   **entry):
    model_schedule, model_field = cls.to_model_schedule(schedule)
    entry.update(
        {model_field: model_schedule},
        args=dumps(args or []),
        kwargs=dumps(kwargs or {}),
        **cls._unpack_options(**options or {})
    )
    return entry

@classmethod
def _unpack_options(cls, queue=None, exchange=None, routing_key=None,
                    **kwargs):
    return {
        'queue': queue,
        'exchange': exchange,
        'routing_key': routing_key,
    }

def __repr__(self):
    return '<ModelEntry: {0} {1}(*{2}, **{3}) {4}>'.format(
        safe_str(self.name), self.task, safe_repr(self.args),
        safe_repr(self.kwargs), self.schedule,
    )

class DatabaseScheduler(Scheduler):
"""Database-backed Beat Scheduler."""

Entry = ModelEntry
Model = PeriodicTask
Changes = PeriodicTasks

_schedule = None
_last_timestamp = None
_initial_read = False

def __init__(self, *args, **kwargs):
    """Initialize the database scheduler."""
    self._dirty = set()
    Scheduler.__init__(self, *args, **kwargs)
    self._finalize = Finalize(self, self.sync, exitpriority=5)
    self.max_interval = (
        kwargs.get('max_interval') or
        self.app.conf.beat_max_loop_interval or
        DEFAULT_MAX_INTERVAL)

def setup_schedule(self):
    self.install_default_entries(self.schedule)
    self.update_from_dict(self.app.conf.beat_schedule)

def all_as_schedule(self):
    debug('DatabaseScheduler: Fetching database schedule')
    s = {}
    for model in self.Model.objects.enabled():
        try:
            s[model.name] = self.Entry(model, app=self.app)
        except ValueError:
            pass
    return s

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).
        try:
            transaction.commit()
        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

def reserve(self, entry):
    new_entry = next(entry)
    # Need to store entry by name, because the entry may change
    # in the mean time.
    self._dirty.add(new_entry.name)
    return new_entry

def sync(self):
    info('Writing entries...')
    _tried = set()
    try:
        close_old_connections()
        with transaction.atomic():
            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 update_from_dict(self, mapping):
    s = {}
    for name, entry_fields in items(mapping):
        try:
            entry = self.Entry.from_entry(name,
                                          app=self.app,
                                          **entry_fields)
            if entry.model.enabled:
                s[name] = entry

        except Exception as exc:
            logger.error(ADD_ENTRY_ERROR, name, exc, entry_fields)
    self.schedule.update(s)

def install_default_entries(self, data):
    entries = {}
    if self.app.conf.result_expires:
        entries.setdefault(
            'celery.backend_cleanup', {
                'task': 'celery.backend_cleanup',
                'schedule': schedules.crontab('0', '4', '*'),
                'options': {'expires': 12 * 3600},
            },
        )
    self.update_from_dict(entries)

@property
def schedule(self):
    update = False
    if not self._initial_read:
        debug('DatabaseScheduler: initial read')
        update = True
        self._initial_read = True
    elif self.schedule_changed():
        info('DatabaseScheduler: Schedule changed.')
        update = True

    if update:
        self.sync()
        self._schedule = self.all_as_schedule()
        # the schedule changed, invalidate the heap in Scheduler.tick
        self._heap = None
        if logger.isEnabledFor(logging.DEBUG):
            debug('Current schedule:\n%s', '\n'.join(
                repr(entry) for entry in values(self._schedule)),
            )
    return self._schedule

[xiaohuazi123]

"""Utilities."""

-- XXX This module must not use translation as that causes

-- a recursive loader import!

from __future__ import absolute_import, unicode_literals

from django.conf import settings
from django.utils import timezone

is_aware = timezone.is_aware

see Issue #222

now_localtime = getattr(timezone, 'template_localtime', timezone.localtime)

def make_aware(value):
"""Force datatime to have timezone information."""
if getattr(settings, 'USE_TZ', False):
# naive datetimes are assumed to be in UTC.
if timezone.is_naive(value):
value = timezone.make_aware(value, timezone.utc)
# then convert to the Django configured timezone.
default_tz = timezone.get_default_timezone()
value = timezone.localtime(value, default_tz)
if not getattr(settings,'CELERY_ENABLE_UTC', True):
if timezone.is_naive(value):
value = value
else:
value = timezone.make_naive(value)
return value

def now():
"""Return the current date and time."""
if getattr(settings, 'USE_TZ', False):
return now_localtime(timezone.now())
else:
return timezone.now()

def is_database_scheduler(scheduler):
"""Return true if Celery is configured to use the db scheduler."""
if not scheduler:
return False
from kombu.utils import symbol_by_name
from .schedulers import DatabaseScheduler
return (
scheduler == 'django' or
issubclass(symbol_by_name(scheduler), DatabaseScheduler)
)

[auvipy]

update with 1.4.0 release?

[thiagoss]

I believe it still happens with 1.4.0.

[Teippo2020]

This bug keeps appearing even on 1.4.0

[niehuan]

I also solved it downgrading to version 1.1.0 and adding to celery.py the following as @zuhao and @kosmos342 said:

app.now = datetime.datetime.now
thx

[jarrod01]

I also solved it downgrading to version 1.1.0 and adding to celery.py the following as @zuhao and @kosmos342 said:

app.now = datetime.datetime.now
thx

thx，me also

[auvipy]

https://github.com/celery/django-celery-beat/pull/231/files anyone tests this patch?

[Teippo2020]

Patch #231 has been merged into master and attempts to fix this issue.

[hulucc]

I'm using 1.5.0, and the problem still.

[vladjkeee13]

You need to add this line to your settings.

DJANGO_CELERY_BEAT_TZ_AWARE = False

[im-n1]

Can anyone confirm ^ solves this issue?

[yaomengsi]

It's solved. @im-n1

[enjoy-binbin]

I'm using 2.0.0, and the problem still.

But i set DJANGO_CELERY_BEAT_TZ_AWARE = False like @vladjkeee13 said

it work for me