def register_periodic_tasks(self, tasks: Iterable[Task]):
        """Register tasks that need to be scheduled periodically."""
        tasks = [task.serialize() for task in tasks]
        self._number_periodic_tasks = len(tasks)
        self._run_script(
            self._register_periodic_tasks,
            math.ceil(datetime.now(timezone.utc).timestamp()),
            self._to_namespaced(PERIODIC_TASKS_HASH_KEY),
            self._to_namespaced(PERIODIC_TASKS_QUEUE_KEY),
            *tasks
        )
def register_periodic_tasks ( self , tasks : Iterable [ Task ] ) : tasks = [ task . serialize ( ) for task in tasks ] self . _number_periodic_tasks = len ( tasks ) self . _run_script ( self . _register_periodic_tasks , math . ceil ( datetime . now ( timezone . utc ) . timestamp ( ) ) , self . _to_namespaced ( PERIODIC_TASKS_HASH_KEY ) , self . _to_namespaced ( PERIODIC_TASKS_QUEUE_KEY ) , * tasks )
Register tasks that need to be scheduled periodically.