celery
diff --git a/‎django_celery_beat/schedulers.py
Lines changed: 142 additions & 40 deletions b/‎django_celery_beat/schedulers.py
Lines changed: 142 additions & 40 deletions
diff --git a/‎django_celery_beat/utils.py
Lines changed: 16 additions & 0 deletions b/‎django_celery_beat/utils.py
Lines changed: 16 additions & 0 deletions
@@ -4,14 +4,20 @@
 import math
 from multiprocessing.util import Finalize
 
+try:
+    from zoneinfo import ZoneInfo  # Python 3.9+
+except ImportError:
+    from backports.zoneinfo import ZoneInfo  # Python 3.8
+
 from celery import current_app, schedules
 from celery.beat import ScheduleEntry, Scheduler
 from celery.utils.log import get_logger
 from celery.utils.time import maybe_make_aware
 from django.conf import settings
 from django.core.exceptions import ObjectDoesNotExist
 from django.db import close_old_connections, transaction
-from django.db.models import Q
+from django.db.models import Case, F, IntegerField, Q, When
+from django.db.models.functions import Cast
 from django.db.utils import DatabaseError, InterfaceError
 from django.utils import timezone
 from kombu.utils.encoding import safe_repr, safe_str
@@ -20,12 +26,13 @@
 from .clockedschedule import clocked
 from .models import (ClockedSchedule, CrontabSchedule, IntervalSchedule,
                      PeriodicTask, PeriodicTasks, SolarSchedule)
-from .utils import NEVER_CHECK_TIMEOUT, now
+from .utils import NEVER_CHECK_TIMEOUT, aware_now, now
 
 # 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
+SCHEDULE_SYNC_MAX_INTERVAL = 300  # 5 minutes
 
 ADD_ENTRY_ERROR = """\
 Cannot add entry %r to database schedule: %r. Contents: %r
@@ -238,6 +245,7 @@ class DatabaseScheduler(Scheduler):
     _last_timestamp = None
     _initial_read = True
     _heap_invalidated = False
+    _last_full_sync = None
 
     def __init__(self, *args, **kwargs):
         """Initialize the database scheduler."""
@@ -256,23 +264,128 @@ def setup_schedule(self):
     def all_as_schedule(self):
         debug('DatabaseScheduler: Fetching database schedule')
         s = {}
-        next_five_minutes = now() + datetime.timedelta(minutes=5)
-        exclude_clock_tasks_query = Q(
-            clocked__isnull=False, clocked__clocked_time__gt=next_five_minutes
+        next_schedule_sync = now() + datetime.timedelta(
+            seconds=SCHEDULE_SYNC_MAX_INTERVAL
         )
-        exclude_hours = self.get_excluded_hours_for_crontab_tasks()
-        exclude_cron_tasks_query = Q(
-            crontab__isnull=False, crontab__hour__in=exclude_hours
+        exclude_clock_tasks_query = Q(
+            clocked__isnull=False,
+            clocked__clocked_time__gt=next_schedule_sync
         )
-        for model in self.Model.objects.enabled().exclude(
-            exclude_clock_tasks_query | exclude_cron_tasks_query
-        ):
+
+        exclude_cron_tasks_query = self._get_crontab_exclude_query()
+
+        # Combine the queries for optimal database filtering
+        exclude_query = exclude_clock_tasks_query | exclude_cron_tasks_query
+
+        # Fetch only the tasks we need to consider
+        for model in self.Model.objects.enabled().exclude(exclude_query):
             try:
                 s[model.name] = self.Entry(model, app=self.app)
             except ValueError:
                 pass
         return s
 
+    def _get_crontab_exclude_query(self):
+        """
+        Build a query to exclude crontab tasks based on their hour value,
+        adjusted for timezone differences relative to the server.
+
+        This creates an annotation for each crontab task that represents the
+        server-equivalent hour, then filters on that annotation.
+        """
+        # Get server time based on Django settings
+
+        server_time = aware_now()
+        server_hour = server_time.hour
+
+        # Window of +/- 2 hours around the current hour in server tz.
+        hours_to_include = [
+            (server_hour + offset) % 24 for offset in range(-2, 3)
+        ]
+        hours_to_include += [4]  # celery's default cleanup task
+
+        # Regex pattern to match only numbers
+        # This ensures we only process numeric hour values
+        numeric_hour_pattern = r'^\d+$'
+
+        # Get all tasks with a simple numeric hour value
+        numeric_hour_tasks = CrontabSchedule.objects.filter(
+            hour__regex=numeric_hour_pattern
+        )
+
+        # Annotate these tasks with their server-hour equivalent
+        annotated_tasks = numeric_hour_tasks.annotate(
+            # Cast hour string to integer
+            hour_int=Cast('hour', IntegerField()),
+
+            # Calculate server-hour based on timezone offset
+            server_hour=Case(
+                # Handle each timezone specifically
+                *[
+                    When(
+                        timezone=timezone_name,
+                        then=(
+                            F('hour_int')
+                            + self._get_timezone_offset(timezone_name)
+                            + 24
+                        ) % 24
+                    )
+                    for timezone_name in self._get_unique_timezone_names()
+                ],
+                # Default case - use hour as is
+                default=F('hour_int')
+            )
+        )
+
+        excluded_hour_task_ids = annotated_tasks.exclude(
+            server_hour__in=hours_to_include
+        ).values_list('id', flat=True)
+
+        # Build the final exclude query:
+        # Exclude crontab tasks that are not in our include list
+        exclude_query = Q(crontab__isnull=False) & Q(
+            crontab__id__in=excluded_hour_task_ids
+        )
+
+        return exclude_query
+
+    def _get_unique_timezone_names(self):
+        """Get a list of all unique timezone names used in CrontabSchedule"""
+        return CrontabSchedule.objects.values_list(
+            'timezone', flat=True
+        ).distinct()
+
+    def _get_timezone_offset(self, timezone_name):
+        """
+        Args:
+            timezone_name: The name of the timezone or a ZoneInfo object
+
+        Returns:
+            int: The hour offset
+        """
+        # Get server timezone
+        server_tz = timezone.get_current_timezone()
+
+        if isinstance(timezone_name, ZoneInfo):
+            timezone_name = timezone_name.key
+
+        target_tz = ZoneInfo(timezone_name)
+
+        # Use a fixed point in time for the calculation to avoid DST issues
+        fixed_dt = datetime.datetime(2023, 1, 1, 12, 0, 0)
+
+        # Calculate the offset
+        dt1 = fixed_dt.replace(tzinfo=server_tz)
+        dt2 = fixed_dt.replace(tzinfo=target_tz)
+
+        # Calculate hour difference
+        offset_seconds = (
+            dt1.utcoffset().total_seconds() - dt2.utcoffset().total_seconds()
+        )
+        offset_hours = int(offset_seconds / 3600)
+
+        return offset_hours
+
     def schedule_changed(self):
         try:
             close_old_connections()
@@ -372,13 +485,31 @@ def schedules_equal(self, *args, **kwargs):
     @property
     def schedule(self):
         initial = update = False
+        current_time = datetime.datetime.now()
+
         if self._initial_read:
             debug('DatabaseScheduler: initial read')
             initial = update = True
             self._initial_read = False
+            self._last_full_sync = current_time
         elif self.schedule_changed():
             info('DatabaseScheduler: Schedule changed.')
             update = True
+            self._last_full_sync = current_time
+
+        # Force update the schedule if it's been more than 5 minutes
+        if not update:
+            time_since_last_sync = (
+                current_time - self._last_full_sync
+            ).total_seconds()
+            if (
+                time_since_last_sync >= SCHEDULE_SYNC_MAX_INTERVAL
+            ):
+                debug(
+                    'DatabaseScheduler: Forcing full sync after 5 minutes'
+                )
+                update = True
+                self._last_full_sync = current_time
 
         if update:
             self.sync()
@@ -392,32 +523,3 @@ def schedule(self):
                     repr(entry) for entry in self._schedule.values()),
                 )
         return self._schedule
-
-    @staticmethod
-    def get_excluded_hours_for_crontab_tasks():
-        # Generate the full list of allowed hours for crontabs
-        allowed_crontab_hours = [
-            f"{hour:02}" for hour in range(24)
-        ] + [
-            str(hour) for hour in range(10)
-        ]
-
-        # Get current, next, and previous hours
-        current_time = timezone.localtime(now())
-        current_hour = current_time.hour
-        next_hour = (current_hour + 1) % 24
-        previous_hour = (current_hour - 1) % 24
-
-        # Create a set of hours to remove (both padded and non-padded versions)
-        hours_to_remove = {
-            f"{current_hour:02}", str(current_hour),
-            f"{next_hour:02}", str(next_hour),
-            f"{previous_hour:02}", str(previous_hour),
-            str(4), "04",  # celery's default cleanup task
-        }
-
-        # Filter out 'should be considered' hours
-        return [
-            hour for hour in allowed_crontab_hours
-            if hour not in hours_to_remove
-        ]
 
@@ -1,8 +1,14 @@
 """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
 
@@ -37,6 +43,16 @@ def now():
         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: