Create or update a Celery Beat periodic task when an EPGSource is created/updated. Skip creating tasks for dummy EPG sources as they don't need refreshing. Supports both interval-based and cron-based scheduling via the shared utility.
(sender, instance, created, update_fields=None, **kwargs)
| 78 | |
| 79 | @receiver(post_save, sender=EPGSource) |
| 80 | def create_or_update_refresh_task(sender, instance, created, update_fields=None, **kwargs): |
| 81 | """ |
| 82 | Create or update a Celery Beat periodic task when an EPGSource is created/updated. |
| 83 | Skip creating tasks for dummy EPG sources as they don't need refreshing. |
| 84 | Supports both interval-based and cron-based scheduling via the shared utility. |
| 85 | """ |
| 86 | # Skip task creation for dummy EPGs |
| 87 | if instance.source_type == 'dummy': |
| 88 | # If there's an existing task, disable it |
| 89 | if instance.refresh_task: |
| 90 | instance.refresh_task.enabled = False |
| 91 | instance.refresh_task.save(update_fields=['enabled']) |
| 92 | return |
| 93 | |
| 94 | # Skip rescheduling when only non-schedule fields were saved (e.g. status/last_message |
| 95 | # updates from the refresh task itself). We only need to reschedule when schedule-relevant |
| 96 | # fields change or when _cron_expression was explicitly set by the serializer. |
| 97 | SCHEDULE_FIELDS = {'refresh_interval', 'is_active', 'refresh_task'} |
| 98 | if ( |
| 99 | not created |
| 100 | and update_fields is not None |
| 101 | and not (set(update_fields) & SCHEDULE_FIELDS) |
| 102 | and not hasattr(instance, '_cron_expression') |
| 103 | ): |
| 104 | return |
| 105 | |
| 106 | task_name = f"epg_source-refresh-{instance.id}" |
| 107 | should_be_enabled = instance.is_active |
| 108 | |
| 109 | # Read cron_expression from transient attribute set by the serializer. |
| 110 | # If not set (e.g. save came from a task updating status/last_message), |
| 111 | # preserve the existing crontab so we don't accidentally revert to interval. |
| 112 | if hasattr(instance, "_cron_expression"): |
| 113 | cron_expr = instance._cron_expression |
| 114 | else: |
| 115 | cron_expr = "" |
| 116 | try: |
| 117 | existing_task = instance.refresh_task |
| 118 | if existing_task and existing_task.crontab: |
| 119 | ct = existing_task.crontab |
| 120 | cron_expr = f"{ct.minute} {ct.hour} {ct.day_of_month} {ct.month_of_year} {ct.day_of_week}" |
| 121 | except Exception: |
| 122 | pass |
| 123 | |
| 124 | task = create_or_update_periodic_task( |
| 125 | task_name=task_name, |
| 126 | celery_task_path="apps.epg.tasks.refresh_epg_data", |
| 127 | kwargs={"source_id": instance.id}, |
| 128 | interval_hours=int(instance.refresh_interval), |
| 129 | cron_expression=cron_expr, |
| 130 | enabled=should_be_enabled, |
| 131 | ) |
| 132 | |
| 133 | if instance.refresh_task != task: |
| 134 | instance.refresh_task = task |
| 135 | instance.save(update_fields=["refresh_task"]) |
| 136 | |
| 137 | @receiver(post_delete, sender=EPGSource) |
nothing calls this directly
no test coverage detected