67 lines
2.1 KiB
Python
67 lines
2.1 KiB
Python
"""Utilities."""
|
|
import datetime
|
|
# -- XXX This module must not use translation as that causes
|
|
# -- a recursive loader import!
|
|
from datetime import timezone as datetime_timezone
|
|
|
|
try:
|
|
from zoneinfo import ZoneInfo # Python 3.9+
|
|
except ImportError:
|
|
from backports.zoneinfo import ZoneInfo # Python 3.8
|
|
|
|
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, datetime_timezone.utc)
|
|
# then convert to the Django configured timezone.
|
|
default_tz = timezone.get_default_timezone()
|
|
value = timezone.localtime(value, default_tz)
|
|
elif timezone.is_naive(value):
|
|
# naive datetimes are assumed to be in local timezone.
|
|
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 aware_now():
|
|
if getattr(settings, 'USE_TZ', True):
|
|
# When USE_TZ is True, return timezone.now()
|
|
return timezone.now()
|
|
else:
|
|
# When USE_TZ is False, use the project's timezone
|
|
project_tz = ZoneInfo(getattr(settings, 'TIME_ZONE', 'UTC'))
|
|
return datetime.datetime.now(project_tz)
|
|
|
|
|
|
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)
|
|
)
|