swh.scheduler.utils module

swh.scheduler.utils.utcnow()

swh.scheduler.utils.get_task(task_name)

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(type: str, policy: Literal['recurring', 'oneshot'], *args, next_run: datetime | None = None, priority: Literal['high', 'normal', 'low'] | None = None, retries_left: int | None = None, **kwargs) → Task

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(origin: ListedOrigin, lister: Lister) → Task

swh.scheduler.utils.create_origin_tasks(origins: List[ListedOrigin], scheduler: SchedulerInterface) → List[Task]

Returns a task dict for each origin, in the same order.

swh.scheduler.utils.create_oneshot_task(type: str, *args, next_run: datetime | None = None, **kwargs) → Task

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())