Delete a PeriodicTask by name and clean up orphaned schedules. Args: task_name: The unique name of the PeriodicTask. Returns: True if a task was found and deleted, False otherwise.
(task_name)
| 149 | |
| 150 | |
| 151 | def delete_periodic_task(task_name): |
| 152 | """ |
| 153 | Delete a PeriodicTask by name and clean up orphaned schedules. |
| 154 | |
| 155 | Args: |
| 156 | task_name: The unique name of the PeriodicTask. |
| 157 | |
| 158 | Returns: |
| 159 | True if a task was found and deleted, False otherwise. |
| 160 | """ |
| 161 | try: |
| 162 | task = PeriodicTask.objects.get(name=task_name) |
| 163 | except PeriodicTask.DoesNotExist: |
| 164 | logger.warning(f"No PeriodicTask found with name '{task_name}'") |
| 165 | return False |
| 166 | |
| 167 | old_interval = task.interval |
| 168 | old_crontab = task.crontab |
| 169 | task_id = task.id |
| 170 | |
| 171 | task.delete() |
| 172 | logger.info(f"Deleted periodic task '{task_name}' (id={task_id})") |
| 173 | |
| 174 | if old_interval: |
| 175 | _cleanup_orphaned_interval(old_interval) |
| 176 | if old_crontab: |
| 177 | _cleanup_orphaned_crontab(old_crontab) |
| 178 | |
| 179 | return True |
| 180 | |
| 181 | |
| 182 | # --------------------------------------------------------------------------- |
no test coverage detected