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_dict(type, policy, *args, **kwargs)

Create a task with type and policy, scheduled for as soon as

possible.

Parameters

• type (str) – Type of oneshot task as per swh-scheduler’s db table task_type’s column (Ex: load-git, check-deposit)

• policy (str) – oneshot or recurring policy

Returns

Expected dictionary for the one-shot task scheduling api (swh.scheduler.backend.create_tasks)

swh.scheduler.utils.create_origin_task_dict(origin: swh.scheduler.model.ListedOrigin, lister: swh.scheduler.model.Lister) → Dict[str, Any]

swh.scheduler.utils.create_origin_task_dicts(origins: List[swh.scheduler.model.ListedOrigin], scheduler: swh.scheduler.interface.SchedulerInterface) → List[Dict[str, Any]]

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

swh.scheduler.utils.create_oneshot_task_dict(type, *args, **kwargs)

Create a oneshot task scheduled for as soon as possible.

Parameters

type (str) – Type of oneshot task as per swh-scheduler’s db table task_type’s column (Ex: load-git, check-deposit)

Returns

Expected dictionary for the one-shot task scheduling api (swh.scheduler.backend.create_tasks)