1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
|
"""Utilities."""
# -- XXX This module must not use translation as that causes
# -- a recursive loader import!
from django.conf import settings
from django.utils import timezone
is_aware = timezone.is_aware
# celery schedstate return None will make it not work
NEVER_CHECK_TIMEOUT = 100000000
# 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)
else:
# naive datetimes are assumed to be in local timezone.
if timezone.is_naive(value):
value = timezone.make_aware(value, timezone.get_default_timezone())
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)
)
|
