Create or update a Celery Beat PeriodicTask. Supports both interval (hours) and cron-based scheduling. When *cron_expression* is provided and non-empty it takes precedence over *interval_hours*. An interval_hours of 0 (with no cron) means the task is disabled. Args:
(
task_name,
celery_task_path,
kwargs=None,
interval_hours=0,
cron_expression="",
enabled=True,
)
| 41 | |
| 42 | |
| 43 | def create_or_update_periodic_task( |
| 44 | task_name, |
| 45 | celery_task_path, |
| 46 | kwargs=None, |
| 47 | interval_hours=0, |
| 48 | cron_expression="", |
| 49 | enabled=True, |
| 50 | ): |
| 51 | """ |
| 52 | Create or update a Celery Beat PeriodicTask. Supports both interval |
| 53 | (hours) and cron-based scheduling. |
| 54 | |
| 55 | When *cron_expression* is provided and non-empty it takes precedence |
| 56 | over *interval_hours*. An interval_hours of 0 (with no cron) means |
| 57 | the task is disabled. |
| 58 | |
| 59 | Args: |
| 60 | task_name: Unique PeriodicTask name. |
| 61 | celery_task_path: Dotted path to the Celery task function. |
| 62 | kwargs: dict of keyword arguments passed to the task. |
| 63 | interval_hours: Interval in hours (0 = disabled when no cron). |
| 64 | cron_expression: 5-part cron string (empty = use interval). |
| 65 | enabled: Whether the task should be enabled. |
| 66 | |
| 67 | Returns: |
| 68 | The PeriodicTask instance (created or updated). |
| 69 | """ |
| 70 | task_kwargs = json.dumps(kwargs or {}) |
| 71 | |
| 72 | # Determine effective enabled state |
| 73 | use_cron = bool(cron_expression and cron_expression.strip()) |
| 74 | should_be_enabled = enabled and (use_cron or interval_hours > 0) |
| 75 | |
| 76 | # Retrieve existing task (if any) to track old schedule objects |
| 77 | old_interval = None |
| 78 | old_crontab = None |
| 79 | try: |
| 80 | existing = PeriodicTask.objects.get(name=task_name) |
| 81 | old_interval = existing.interval |
| 82 | old_crontab = existing.crontab |
| 83 | except PeriodicTask.DoesNotExist: |
| 84 | existing = None |
| 85 | |
| 86 | if use_cron: |
| 87 | # ---- Cron-based schedule ---- |
| 88 | cron_parts = parse_cron_expression(cron_expression) |
| 89 | system_tz = CoreSettings.get_system_time_zone() |
| 90 | |
| 91 | crontab, _ = CrontabSchedule.objects.get_or_create( |
| 92 | minute=cron_parts["minute"], |
| 93 | hour=cron_parts["hour"], |
| 94 | day_of_week=cron_parts["day_of_week"], |
| 95 | day_of_month=cron_parts["day_of_month"], |
| 96 | month_of_year=cron_parts["month_of_year"], |
| 97 | timezone=system_tz, |
| 98 | ) |
| 99 | |
| 100 | defaults = { |
no test coverage detected