swh.scheduler.utils module#
- swh.scheduler.utils.get_task(task_name)[source]#
Retrieve task object in our application instance by its fully qualified python name.
- Parameters:
task_name (str) – task’s name (e.g swh.loader.git.tasks.LoadDiskGitRepository)
- Returns:
Instance of task
- swh.scheduler.utils.create_task_dict(type: str, policy: str, *args, next_run: Optional[datetime] = None, **kwargs) Dict[str, Any] [source]#
- Create a task with type and policy, scheduled for as soon as
possible.
- Parameters:
type – Type of oneshot task as per swh-scheduler’s db table task_type’s column (Ex: load-git, check-deposit)
policy – oneshot or recurring policy
next_run – optional date and time from which the task can be executed, use current time otherwise
- Returns:
Expected dictionary for the one-shot task scheduling api (swh.scheduler.backend.create_tasks)
- swh.scheduler.utils.create_origin_task_dict(origin: ListedOrigin, lister: Lister) Dict[str, Any] [source]#
- swh.scheduler.utils.create_origin_task_dicts(origins: List[ListedOrigin], scheduler: SchedulerInterface) List[Dict[str, Any]] [source]#
Returns a task dict for each origin, in the same order.
- swh.scheduler.utils.create_oneshot_task_dict(type: str, *args, next_run: Optional[datetime] = None, **kwargs)[source]#
Create a oneshot task scheduled for as soon as possible.
- Parameters:
type – Type of oneshot task as per swh-scheduler’s db table task_type’s column (Ex: load-git, check-deposit)
next_run – optional date and time from which the task can be executed, use current time otherwise
- Returns:
Expected dictionary for the one-shot task scheduling api (
swh.scheduler.backend.create_tasks()
)