How to delete scheduled tasks

[casper-71]

Hi guys!

I have dynamic scheduled tasks in my FastAPI project, and the project consists of the following steps:

1. Creating an object
2. Updating the object

When creating an object: task.schedule_by_time(time=<some_time>), but when attempting to update the object, the schedule time should also be updated.

How do I do this?
Thanks.

[danfimov]

After calling schedule_by_time as a result you will get CreatedSchedule object like this (code source):

classCreatedSchedule(Generic[_ReturnType]):
"""A schedule that has been created."""def__init__(
self,
kicker: "AsyncKicker[Any,_ReturnType]",
source: ScheduleSource,
task: ScheduledTask,
) ->None:
self.kicker=kickerself.source=sourceself.task=taskself.schedule_id=task.schedule_id

To update schedule time you can use fields from this object to delete previous schedule and add a new one with different time:

# It's a pseudocode to demonstrate this idea:# 1) Create schedule source schedule_source= ...
awaitschedule_source.startup()
# 2) Schedule task for the first timecreated_schedule=awaitmy_task.schedule_by_time(
schedule_source,
datetime.datetime.now(datetime.UTC) +datetime.timedelta(minutes=1, seconds=0),
)
# 3) Reschedule taskawaitschedule_source.delete_schedule(created_schedule.schedule_id)
new_scheduled_task=created_schedule.tasknew_scheduled_task.time=datetime.datetime.now(datetime.UTC) +datetime.timedelta(minutes=2, seconds=0)
awaitschedule_source.add_schedule(new_scheduled_task)

This methods of schedule source are optional according to documentation, but they are implemented in most of them.